diff --git a/apps/worker/video_processing/render_adapter.py b/apps/worker/video_processing/render_adapter.py
old mode 100755
new mode 100644
index 8d84f9735..6b854514c
--- a/apps/worker/video_processing/render_adapter.py
+++ b/apps/worker/video_processing/render_adapter.py
@@ -531,6 +531,23 @@ class RenderAdapter:
         )
         result = render_svc.render()
 
+        # 4.5 渲染后校验输出完整性
+        from video_processing.video_validation import validate_video_output
+
+        validation = validate_video_output(result.output_path)
+        if not validation.valid:
+            logger.error(
+                "[render-adapter] 渲染输出校验失败: plan_id=%s job_id=%s error=%s",
+                plan_id,
+                job_id,
+                validation.error_message,
+            )
+            return RenderAdapterResult(
+                success=False,
+                error_message=f"渲染输出校验失败: {validation.error_message}",
+                error_detail=validation.error_message,
+            )
+
         self._report_progress(progress_cb, 80.0, "上传渲染结果")
 
         # 5. 上传结果
diff --git a/apps/worker/video_processing/video_validation.py b/apps/worker/video_processing/video_validation.py
new file mode 100755
index 000000000..9659edfc2
--- /dev/null
+++ b/apps/worker/video_processing/video_validation.py
@@ -0,0 +1,267 @@
+"""视频输出校验 — 渲染后验证输出文件完整性.
+
+在 FFmpeg 渲染完成后，验证输出文件是否有效（非损坏/截断）：
+  1. 文件存在且大小 > 0
+  2. moov atom 存在（MP4 容器完整性标志）
+  3. ffprobe 可正常读取视频流信息
+
+用途：
+  - RenderAdapter 渲染后调用，避免将损坏文件上传到 OSS
+  - 提前发现 FFmpeg 异常退出但未抛异常的情况（如 exit=0 但输出截断）
+"""
+
+from __future__ import annotations
+
+import logging
+import subprocess
+from dataclasses import dataclass
+from pathlib import Path
+
+from shared.ffmpeg_utils import FFPROBE_BIN
+
+logger = logging.getLogger(__name__)
+
+
+# ── 数据结构 ──────────────────────────────────────────────────────────────────
+
+
+@dataclass
+class VideoValidationResult:
+    """视频校验结果。"""
+
+    valid: bool
+    file_exists: bool = False
+    file_size: int = 0
+    moov_atom_found: bool = False
+    has_video_stream: bool = False
+    duration: float = 0.0
+    width: int = 0
+    height: int = 0
+    error_message: str = ""
+
+    @property
+    def is_valid(self) -> bool:
+        """是否通过所有校验。"""
+        return self.valid
+
+
+# ── FFmpeg 退出码映射 ──────────────────────────────────────────────────────────
+
+# 常见 FFmpeg 退出码及其含义
+FFMPEG_EXIT_CODES: dict[int, tuple[str, str]] = {
+    0: ("成功", "渲染正常完成"),
+    1: ("通用错误", "FFmpeg 执行出错，请检查输入参数和素材"),
+    69: ("权限错误", "无权限写入输出文件或访问输入文件"),
+    126: ("权限不足", "命令不可执行"),
+    127: ("命令不存在", "FFmpeg 二进制文件未找到"),
+    134: ("Abnormal termination", "FFmpeg 异常终止（可能内存不足）"),
+    137: ("OOM Killed", "FFmpeg 被系统 OOM Killer 终止（内存不足）"),
+    139: ("段错误", "FFmpeg Segmentation Fault（可能是编解码器 Bug）"),
+    141: ("管道断裂", "FFmpeg 输出管道断裂"),
+    143: ("SIGTERM", "FFmpeg 收到终止信号"),
+    183: ("滤镜错误", "FFmpeg 滤镜链配置错误"),
+    234: ("素材异常", "输入素材格式不兼容或已损坏"),
+    255: ("严重错误", "FFmpeg 执行严重错误"),
+}
+
+
+def get_exit_code_message(exit_code: int) -> str:
+    """获取 FFmpeg 退出码的中文描述。
+
+    Args:
+        exit_code: FFmpeg 进程退出码
+
+    Returns:
+        人类可读的错误描述
+    """
+    if exit_code in FFMPEG_EXIT_CODES:
+        name, desc = FFMPEG_EXIT_CODES[exit_code]
+        return f"exit={exit_code} ({name}): {desc}"
+    if exit_code > 128:
+        signal_num = exit_code - 128
+        return f"exit={exit_code}: 被信号 {signal_num} 终止"
+    return f"exit={exit_code}: 未知错误"
+
+
+# ── 校验函数 ──────────────────────────────────────────────────────────────────
+
+
+def validate_video_output(video_path: str | Path, *, min_duration: float = 0.1) -> VideoValidationResult:
+    """校验渲染输出视频文件的完整性。
+
+    校验项：
+      1. 文件存在且大小 > 0
+      2. 包含 moov atom（MP4 容器完整性）
+      3. ffprobe 可读取至少一个视频流
+
+    Args:
+        video_path: 输出视频文件路径
+        min_duration: 最小有效时长（秒），低于此值视为无效，默认 0.1s
+
+    Returns:
+        VideoValidationResult
+    """
+    path = Path(video_path)
+    result = VideoValidationResult(valid=False)
+
+    # 1. 文件存在性检查
+    if not path.exists():
+        result.error_message = f"输出文件不存在: {path}"
+        logger.error("[video-validation] %s", result.error_message)
+        return result
+
+    result.file_exists = True
+
+    # 2. 文件大小检查
+    try:
+        result.file_size = path.stat().st_size
+    except OSError as e:
+        result.error_message = f"无法读取文件大小: {e}"
+        logger.error("[video-validation] %s", result.error_message)
+        return result
+
+    if result.file_size == 0:
+        result.error_message = "输出文件大小为 0（FFmpeg 未写入任何数据）"
+        logger.error("[video-validation] %s", result.error_message)
+        return result
+
+    # 极小文件（< 1KB）几乎不可能是有效视频
+    if result.file_size < 1024:
+        result.error_message = f"输出文件过小 ({result.file_size} bytes)，可能渲染未完成"
+        logger.error("[video-validation] %s", result.error_message)
+        return result
+
+    # 3. moov atom 检查（MP4 容器完整性标志）
+    result.moov_atom_found = _check_moov_atom(path)
+    if not result.moov_atom_found:
+        result.error_message = (
+            "输出文件缺少 moov atom（MP4 容器不完整）。" "可能原因：FFmpeg 被强制终止、磁盘空间不足、或渲染过程中断。"
+        )
+        logger.error("[video-validation] %s — file_size=%d", result.error_message, result.file_size)
+        return result
+
+    # 4. ffprobe 验证视频流
+    try:
+        probe_result = subprocess.run(  # nosec B603
+            [
+                FFPROBE_BIN,
+                "-v",
+                "error",
+                "-select_streams",
+                "v:0",
+                "-show_entries",
+                "stream=width,height,duration,codec_name",
+                "-show_entries",
+                "format=duration",
+                "-of",
+                "json",
+                str(path),
+            ],
+            check=True,
+            stdout=subprocess.PIPE,
+            stderr=subprocess.PIPE,
+            text=True,
+            timeout=15,
+        )
+
+        import json
+
+        probe_data = json.loads(probe_result.stdout)
+        streams = probe_data.get("streams", [])
+        fmt = probe_data.get("format", {})
+
+        if not streams:
+            result.error_message = "输出文件无视频流（ffprobe 未检测到 video stream）"
+            logger.error("[video-validation] %s", result.error_message)
+            return result
+
+        result.has_video_stream = True
+
+        video_stream = streams[0]
+        result.width = int(video_stream.get("width", 0) or 0)
+        result.height = int(video_stream.get("height", 0) or 0)
+        result.duration = float(fmt.get("duration", 0) or video_stream.get("duration", 0) or 0)
+
+        # 时长校验
+        if result.duration > 0 and result.duration < min_duration:
+            result.error_message = (
+                f"输出视频时长过短 ({result.duration:.2f}s < {min_duration}s)，" "可能渲染只处理了极少帧"
+            )
+            logger.warning("[video-validation] %s", result.error_message)
+            # 不标记为失败，只是警告（某些预览场景确实很短）
+            # 但如果时长为 0 且文件大小较大，说明 moov 有问题
+        elif result.duration == 0 and result.file_size > 0:
+            logger.warning(
+                "[video-validation] ffprobe 无法读取时长，但文件存在且大小=%d，标记为可疑",
+                result.file_size,
+            )
+
+    except subprocess.TimeoutExpired:
+        result.error_message = "ffprobe 超时（15s），输出文件可能已损坏"
+        logger.error("[video-validation] %s", result.error_message)
+        return result
+    except subprocess.CalledProcessError as e:
+        stderr_text = (e.stderr or "").strip()
+        result.error_message = f"ffprobe 校验失败: exit={e.returncode}, stderr={stderr_text[:200]}"
+        logger.error("[video-validation] %s", result.error_message)
+        return result
+    except (json.JSONDecodeError, KeyError, ValueError) as e:
+        result.error_message = f"ffprobe 输出解析失败: {e}"
+        logger.error("[video-validation] %s", result.error_message)
+        return result
+
+    # 全部通过
+    result.valid = True
+    logger.info(
+        "[video-validation] 校验通过: path=%s size=%d duration=%.2fs resolution=%dx%d",
+        path,
+        result.file_size,
+        result.duration,
+        result.width,
+        result.height,
+    )
+    return result
+
+
+def _check_moov_atom(video_path: Path) -> bool:
+    """检查 MP4 文件是否包含 moov atom。
+
+    moov atom 是 MP4 容器的元数据容器，包含视频时长、编解码器信息等。
+    FFmpeg 使用 -movflags +faststart 时 moov 在文件头部；否则在尾部。
+    如果 FFmpeg 被强制终止，moov 可能完全不存在。
+
+    方法：读取文件前 64KB + 尾部 64KB，搜索 "moov" 字节标记。
+    对于 -movflags +faststart 的快启文件，moov 在头部。
+
+    Args:
+        video_path: 视频文件路径
+
+    Returns:
+        True 表示找到 moov atom
+    """
+    try:
+        file_size = video_path.stat().st_size
+        if file_size < 8:
+            return False
+
+        # 搜索范围：头部 64KB + 尾部 64KB（覆盖 faststart 和普通 MP4）
+        search_size = min(64 * 1024, file_size)
+
+        with open(video_path, "rb") as f:
+            # 读取头部
+            head_data = f.read(search_size)
+            if b"moov" in head_data:
+                return True
+
+            # 读取尾部
+            if file_size > search_size:
+                f.seek(file_size - search_size)
+                tail_data = f.read(search_size)
+                if b"moov" in tail_data:
+                    return True
+
+        return False
+
+    except OSError as e:
+        logger.warning("[video-validation] 检查 moov atom 失败: %s", e)
+        return False
diff --git a/apps/worker/worker_app/tasks/_startup.py b/apps/worker/worker_app/tasks/_startup.py
old mode 100644
new mode 100755
index 2a66fc74d..147983330
--- a/apps/worker/worker_app/tasks/_startup.py
+++ b/apps/worker/worker_app/tasks/_startup.py
@@ -1,4 +1,4 @@
-"""Worker 启动时的初始化任务 — 孤儿任务清理等。"""
+"""Worker 启动时的初始化任务 — 孤儿任务清理等."""
 
 import logging
 
@@ -7,11 +7,12 @@ from worker_app.db import SessionLocal
 
 logger = logging.getLogger(__name__)
 
+# 孤儿任务超时阈值：渲染任务超过此时间未更新则视为卡死
 ORPHAN_TASK_TIMEOUT_MINUTES = 10
 
 
 def cleanup_orphan_tasks(timeout_minutes: int = ORPHAN_TASK_TIMEOUT_MINUTES) -> int:
-    """清理数据库中超时未更新的 running 任务。
+    """清理数据库中超时未更新的 running GenerationTask（孤儿任务）。
 
     worker 重启或崩溃后，之前处于 running 状态的任务会变成孤儿任务，
     一直卡在 running 不动。通过 updated_at 超时判断并标记为 failed。
@@ -32,18 +33,81 @@ def cleanup_orphan_tasks(timeout_minutes: int = ORPHAN_TASK_TIMEOUT_MINUTES) ->
         count = repo.cleanup_stale_running(timeout_minutes)
         session.close()
         if count > 0:
-            logger.warning("清理了 %d 个超时的孤儿 running 任务", count)
+            logger.warning("清理了 %d 个超时的孤儿 GenerationTask（超过 %d 分钟未更新）", count, timeout_minutes)
         else:
-            logger.info("无孤儿 running 任务需要清理")
+            logger.info("无孤儿 GenerationTask 需要清理")
         return count
     except Exception as e:
-        logger.error("清理孤儿任务失败: %s", e, exc_info=True)
+        logger.error("清理孤儿 GenerationTask 失败: %s", e, exc_info=True)
         return 0
 
 
+def cleanup_stale_jobs(timeout_minutes: int = ORPHAN_TASK_TIMEOUT_MINUTES) -> int:
+    """清理数据库中超时未更新的 running Job（孤儿任务）。
+
+    与 cleanup_orphan_tasks 配合，同时清理 Job 表和 GenerationTask 表。
+
+    Returns:
+        清理的任务数量
+    """
+    from datetime import datetime, timedelta, timezone
+
+    from packages.adapters.sqlalchemy_impl.models import JobModel
+    from packages.domain.job import JobStatus
+
+    try:
+        session = SessionLocal()
+        cutoff = datetime.now(timezone.utc) - timedelta(minutes=timeout_minutes)
+        stale_jobs = (
+            session.query(JobModel)
+            .filter(
+                JobModel.status == JobStatus.RUNNING.value,
+                JobModel.updated_at < cutoff,
+            )
+            .all()
+        )
+        count = 0
+        for model in stale_jobs:
+            model.status = JobStatus.FAILED.value
+            model.error_message = f"任务执行中断（超过 {timeout_minutes} 分钟未更新）"
+            count += 1
+        if count > 0:
+            session.commit()
+            logger.warning("清理了 %d 个超时的孤儿 Job（超过 %d 分钟未更新）", count, timeout_minutes)
+        else:
+            logger.info("无孤儿 Job 需要清理")
+        session.close()
+        return count
+    except Exception as e:
+        logger.error("清理孤儿 Job 失败: %s", e, exc_info=True)
+        return 0
+
+
+def cleanup_all_stale_tasks(timeout_minutes: int = ORPHAN_TASK_TIMEOUT_MINUTES) -> dict:
+    """统一清理所有超时的孤儿任务。
+
+    同时清理 GenerationTask 和 Job 两类表。
+
+    Returns:
+        {"generation_tasks": int, "jobs": int}
+    """
+    gen_count = cleanup_orphan_tasks(timeout_minutes)
+    job_count = cleanup_stale_jobs(timeout_minutes)
+    total = gen_count + job_count
+    if total > 0:
+        logger.warning(
+            "孤儿任务清理完成: GenerationTask=%d, Job=%d, 总计=%d",
+            gen_count,
+            job_count,
+            total,
+        )
+    return {"generation_tasks": gen_count, "jobs": job_count}
+
+
 @worker_ready.connect
 def _on_worker_ready(sender, **kwargs):
     """Worker 启动完成后执行 — 清理孤儿任务。"""
     logger.info("Worker 启动完成，开始清理孤儿 running 任务...")
-    count = cleanup_orphan_tasks()
-    logger.info("Worker 启动清理完成，共清理 %d 个孤儿任务", count)
+    result = cleanup_all_stale_tasks()
+    total = result["generation_tasks"] + result["jobs"]
+    logger.info("Worker 启动清理完成，共清理 %d 个孤儿任务", total)
diff --git a/apps/worker/worker_app/tasks/compose_video.py b/apps/worker/worker_app/tasks/compose_video.py
index f4b706395..1531f548f 100755
--- a/apps/worker/worker_app/tasks/compose_video.py
+++ b/apps/worker/worker_app/tasks/compose_video.py
@@ -6,15 +6,22 @@
 from __future__ import annotations
 
 import os
+import subprocess
 import tempfile
 from pathlib import Path
 
+from celery.exceptions import SoftTimeLimitExceeded
 from celery.utils.log import get_task_logger
 from worker_app.celery_app import celery_app
 from worker_app.db import SessionLocal
 
 logger = get_task_logger(__name__)
 
+# 任务超时时间（秒）：超过此时间 Celery 会抛出 SoftTimeLimitExceeded
+RENDER_TASK_SOFT_TIME_LIMIT = 600  # 10 分钟
+# 硬超时：超过此时间进程会被强制 kill
+RENDER_TASK_TIME_LIMIT = 660  # 10 分钟 + 1 分钟清理缓冲
+
 
 def _get_job_service():
     """延迟导入 JobService，避免循环依赖。"""
@@ -31,6 +38,8 @@ def _get_job_service():
     bind=True,
     max_retries=3,
     default_retry_delay=60,
+    soft_time_limit=RENDER_TASK_SOFT_TIME_LIMIT,
+    time_limit=RENDER_TASK_TIME_LIMIT,
 )
 def compose_video(self, job_id: str, **kwargs):
     """视频合成任务。
@@ -57,9 +66,51 @@ def compose_video(self, job_id: str, **kwargs):
         # 使用 unified 渲染引擎
         return _compose_with_unified_engine(self, job_service, job, plan_id, db)
 
+    except SoftTimeLimitExceeded:
+        # Celery 软超时：任务执行超过 soft_time_limit
+        error_msg = f"渲染任务超时（超过 {RENDER_TASK_SOFT_TIME_LIMIT // 60} 分钟）"
+        logger.error("视频合成超时: job_id=%s", job_id)
+        try:
+            job_service.fail_job(job_id, error_msg)
+        except Exception:
+            logger.exception("更新 Job 超时失败状态时出错")
+        # 超时不重试
+        return {"status": "error", "message": error_msg, "error_type": "timeout"}
+
+    except subprocess.TimeoutExpired as exc:
+        # FFmpeg 子进程超时
+        error_msg = f"FFmpeg 渲染超时（{exc.timeout}s）"
+        logger.error("视频合成 FFmpeg 超时: job_id=%s timeout=%s", job_id, exc.timeout)
+        try:
+            job_service.fail_job(job_id, error_msg)
+        except Exception:
+            logger.exception("更新 Job 超时失败状态时出错")
+        # 超时不重试
+        return {"status": "error", "message": error_msg, "error_type": "ffmpeg_timeout"}
+
     except self.retry_exc as exc:
         logger.warning("视频合成重试中: job_id=%s, exc=%s", job_id, exc)
         raise
+
+    except subprocess.CalledProcessError as exc:
+        # FFmpeg 执行失败，提取有意义的错误信息
+        from video_processing.video_validation import get_exit_code_message
+
+        exit_msg = get_exit_code_message(exc.returncode)
+        stderr_text = (exc.stderr or "").strip()
+        stderr_tail = stderr_text[-300:] if len(stderr_text) > 300 else stderr_text
+        error_msg = f"渲染失败: {exit_msg}"
+        if stderr_tail:
+            error_msg += f" | {stderr_tail[:200]}"
+
+        logger.error("视频合成 FFmpeg 失败: job_id=%s %s", job_id, exit_msg)
+        try:
+            job_service.fail_job(job_id, error_msg[:500])
+        except Exception:
+            logger.exception("更新 Job 失败状态时出错")
+        # FFmpeg 错误不重试（通常是素材或配置问题）
+        return {"status": "error", "message": error_msg, "error_type": "ffmpeg_error", "exit_code": exc.returncode}
+
     except Exception as exc:
         logger.exception("视频合成异常: job_id=%s", job_id)
         try:
@@ -108,7 +159,8 @@ def _compose_with_unified_engine(task, job_service, job, plan_id: str, db) -> di
     )
 
     if not result.success:
-        job_service.fail_job(job_id, f"渲染失败: {result.error_message}")
+        error_msg = f"渲染失败: {result.error_message}"
+        job_service.fail_job(job_id, error_msg[:500])
         raise RuntimeError(result.error_message)
 
     # 更新 Job 状态为完成
diff --git a/apps/worker/worker_app/tasks/edit_plan_generation.py b/apps/worker/worker_app/tasks/edit_plan_generation.py
old mode 100755
new mode 100644
index c1f52ecac..94020f6de
--- a/apps/worker/worker_app/tasks/edit_plan_generation.py
+++ b/apps/worker/worker_app/tasks/edit_plan_generation.py
@@ -271,7 +271,13 @@ def _render_with_unified(
     )
 
 
-@celery_app.task(name="worker.render_edit_plan", bind=True, max_retries=2)
+@celery_app.task(
+    name="worker.render_edit_plan",
+    bind=True,
+    max_retries=2,
+    soft_time_limit=600,  # 10 分钟软超时
+    time_limit=660,  # 11 分钟硬超时
+)
 def render_edit_plan(self, plan_id: str) -> dict:
     """渲染剪辑计划
 
@@ -360,8 +366,17 @@ def render_edit_plan(self, plan_id: str) -> dict:
             return result
 
         except Exception as exc:
-            logger.exception("渲染剪辑计划异常: %s", plan_id)
+            # 超时异常不重试，直接标记失败
+            from celery.exceptions import SoftTimeLimitExceeded
+
+            is_timeout = isinstance(exc, SoftTimeLimitExceeded)
+            if is_timeout:
+                logger.error("渲染剪辑计划超时: plan_id=%s", plan_id)
+            else:
+                logger.exception("渲染剪辑计划异常: %s", plan_id)
+
             # 尝试标记计划和 GenerationTask 为失败
+            error_msg = "渲染任务超时（超过10分钟）" if is_timeout else f"渲染异常: {type(exc).__name__}: {exc}"
             try:
                 plan = plan_repo.get(plan_id)
                 if plan and plan.status.value == "rendering":
@@ -375,12 +390,13 @@ def render_edit_plan(self, plan_id: str) -> dict:
                     gen_task = gen_task_repo.get(generation_task_id)
                     if gen_task and gen_task.status.value != "failed":
                         gen_task.status = "failed"
-                        gen_task.error_message = f"渲染异常: {type(exc).__name__}: {exc}"
+                        gen_task.error_message = error_msg
                         gen_task.completed_at = datetime.now(timezone.utc)
                         try:
+                            log_stage = "render_timeout" if is_timeout else "render_failed"
                             gen_task.append_log(
-                                stage="render_failed",
-                                message=f"渲染异常: {type(exc).__name__}: {str(exc)[:500]}",
+                                stage=log_stage,
+                                message=error_msg[:500],
                                 level="ERROR",
                                 exception_type=type(exc).__name__,
                             )
diff --git a/apps/worker/worker_app/tasks/generation.py b/apps/worker/worker_app/tasks/generation.py
old mode 100755
new mode 100644
index 6a7092671..c28eb29b4
--- a/apps/worker/worker_app/tasks/generation.py
+++ b/apps/worker/worker_app/tasks/generation.py
@@ -1206,7 +1206,13 @@ def _upload_and_record(
 # ── Celery Task ──────────────────────────────────────────────────────────────
 
 
-@celery_app.task(bind=True, name="worker.generate_video", max_retries=2)
+@celery_app.task(
+    bind=True,
+    name="worker.generate_video",
+    max_retries=2,
+    soft_time_limit=600,  # 10 分钟软超时
+    time_limit=660,  # 11 分钟硬超时
+)
 def generate_video(self, task_id: str) -> dict:
     """生成视频任务 — 使用 UnifiedRenderService 统一渲染。
 
diff --git a/tests/unit/test_video_validation.py b/tests/unit/test_video_validation.py
new file mode 100755
index 000000000..73494241c
--- /dev/null
+++ b/tests/unit/test_video_validation.py
@@ -0,0 +1,334 @@
+"""测试 video_validation 模块 — 渲染输出校验."""
+
+from __future__ import annotations
+
+import json
+import subprocess
+import tempfile
+from pathlib import Path
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from video_processing.video_validation import (
+    FFMPEG_EXIT_CODES,
+    VideoValidationResult,
+    _check_moov_atom,
+    get_exit_code_message,
+    validate_video_output,
+)
+
+# ── get_exit_code_message 测试 ────────────────────────────────────────────────
+
+
+class TestGetExitCodeMessage:
+    """FFmpeg 退出码映射测试."""
+
+    def test_known_exit_codes(self):
+        """已知退出码返回有意义的描述."""
+        assert "成功" in get_exit_code_message(0)
+        assert "通用错误" in get_exit_code_message(1)
+        assert "OOM" in get_exit_code_message(137)
+        assert "段错误" in get_exit_code_message(139)
+        assert "滤镜错误" in get_exit_code_message(183)
+        assert "素材异常" in get_exit_code_message(234)
+
+    def test_signal_termination(self):
+        """信号终止（exit > 128 且不在映射表中）返回信号编号."""
+        msg = get_exit_code_message(130)  # SIGINT = 130 - 128 = 2
+        assert "信号 2" in msg
+
+    def test_unknown_exit_code(self):
+        """未知退出码返回通用错误."""
+        msg = get_exit_code_message(42)
+        assert "未知错误" in msg
+        assert "42" in msg
+
+    def test_all_mapped_codes_have_description(self):
+        """所有映射的退出码都有名称和描述."""
+        for code, (name, desc) in FFMPEG_EXIT_CODES.items():
+            assert name, f"exit_code {code} 缺少名称"
+            assert desc, f"exit_code {code} 缺少描述"
+
+
+# ── _check_moov_atom 测试 ─────────────────────────────────────────────────────
+
+
+class TestCheckMoovAtom:
+    """moov atom 检测测试."""
+
+    def test_file_with_moov_in_head(self):
+        """文件头部包含 moov 标记 → True（faststart 模式）."""
+        with tempfile.NamedTemporaryFile(suffix=".mp4", delete=False) as f:
+            # 写入一些假数据，中间包含 moov 标记
+            f.write(b"\x00" * 100)
+            f.write(b"moov")
+            f.write(b"\x00" * 1000)
+            f.flush()
+            path = Path(f.name)
+
+        try:
+            assert _check_moov_atom(path) is True
+        finally:
+            path.unlink()
+
+    def test_file_with_moov_in_tail(self):
+        """文件尾部包含 moov 标记 → True（普通 MP4 模式）."""
+        with tempfile.NamedTemporaryFile(suffix=".mp4", delete=False) as f:
+            # 写一个较大的文件，moov 在尾部
+            f.write(b"\x00" * 100_000)
+            f.write(b"moov")
+            f.write(b"\x00" * 100)
+            f.flush()
+            path = Path(f.name)
+
+        try:
+            assert _check_moov_atom(path) is True
+        finally:
+            path.unlink()
+
+    def test_file_without_moov(self):
+        """文件不含 moov 标记 → False（截断/损坏的 MP4）."""
+        with tempfile.NamedTemporaryFile(suffix=".mp4", delete=False) as f:
+            # 写入纯随机数据，不含 moov
+            f.write(b"\x00\x01\x02\x03" * 1000)
+            f.flush()
+            path = Path(f.name)
+
+        try:
+            assert _check_moov_atom(path) is False
+        finally:
+            path.unlink()
+
+    def test_empty_file(self):
+        """空文件 → False."""
+        with tempfile.NamedTemporaryFile(suffix=".mp4", delete=False) as f:
+            path = Path(f.name)
+
+        try:
+            assert _check_moov_atom(path) is False
+        finally:
+            path.unlink()
+
+    def test_tiny_file(self):
+        """极小文件（< 8 bytes）→ False."""
+        with tempfile.NamedTemporaryFile(suffix=".mp4", delete=False) as f:
+            f.write(b"\x00\x01")
+            f.flush()
+            path = Path(f.name)
+
+        try:
+            assert _check_moov_atom(path) is False
+        finally:
+            path.unlink()
+
+    def test_nonexistent_file(self):
+        """不存在的文件 → False."""
+        assert _check_moov_atom(Path("/nonexistent/file.mp4")) is False
+
+
+# ── validate_video_output 测试 ────────────────────────────────────────────────
+
+
+class TestValidateVideoOutput:
+    """渲染输出完整性校验测试."""
+
+    def test_file_not_exists(self):
+        """文件不存在 → valid=False."""
+        result = validate_video_output("/nonexistent/video.mp4")
+        assert result.valid is False
+        assert result.file_exists is False
+        assert "不存在" in result.error_message
+
+    def test_empty_file(self):
+        """空文件 → valid=False."""
+        with tempfile.NamedTemporaryFile(suffix=".mp4", delete=False) as f:
+            path = Path(f.name)
+
+        try:
+            result = validate_video_output(path)
+            assert result.valid is False
+            assert "大小" in result.error_message or "过小" in result.error_message
+        finally:
+            path.unlink()
+
+    def test_tiny_file(self):
+        """极小文件（< 1KB）→ valid=False."""
+        with tempfile.NamedTemporaryFile(suffix=".mp4", delete=False) as f:
+            f.write(b"\x00" * 500)
+            path = Path(f.name)
+
+        try:
+            result = validate_video_output(path)
+            assert result.valid is False
+            assert "过小" in result.error_message
+        finally:
+            path.unlink()
+
+    def test_file_without_moov(self):
+        """文件有大小但无 moov atom → valid=False."""
+        with tempfile.NamedTemporaryFile(suffix=".mp4", delete=False) as f:
+            f.write(b"\x00\x01\x02\x03" * 2000)  # 8KB, 无 moov
+            path = Path(f.name)
+
+        try:
+            result = validate_video_output(path)
+            assert result.valid is False
+            assert "moov" in result.error_message.lower()
+        finally:
+            path.unlink()
+
+    @patch("video_processing.video_validation.subprocess.run")
+    def test_valid_video(self, mock_run):
+        """完整的有效视频 → valid=True."""
+        # mock ffprobe 返回
+        mock_result = MagicMock()
+        mock_result.stdout = json.dumps(
+            {
+                "streams": [
+                    {
+                        "width": 1080,
+                        "height": 1920,
+                        "duration": "10.5",
+                        "codec_name": "h264",
+                    }
+                ],
+                "format": {"duration": "10.5"},
+            }
+        )
+        mock_result.returncode = 0
+        mock_run.return_value = mock_result
+
+        # 创建含 moov 的文件
+        with tempfile.NamedTemporaryFile(suffix=".mp4", delete=False) as f:
+            f.write(b"\x00" * 100)
+            f.write(b"moov")
+            f.write(b"\x00" * 5000)
+            path = Path(f.name)
+
+        try:
+            result = validate_video_output(path)
+            assert result.valid is True
+            assert result.file_exists is True
+            assert result.moov_atom_found is True
+            assert result.has_video_stream is True
+            assert result.width == 1080
+            assert result.height == 1920
+            assert result.duration == 10.5
+            assert result.error_message == ""
+        finally:
+            path.unlink()
+
+    @patch("video_processing.video_validation.subprocess.run")
+    def test_no_video_stream(self, mock_run):
+        """文件有 moov 但无视频流 → valid=False."""
+        mock_result = MagicMock()
+        mock_result.stdout = json.dumps({"streams": [], "format": {"duration": "0"}})
+        mock_result.returncode = 0
+        mock_run.return_value = mock_result
+
+        with tempfile.NamedTemporaryFile(suffix=".mp4", delete=False) as f:
+            f.write(b"\x00" * 100)
+            f.write(b"moov")
+            f.write(b"\x00" * 5000)
+            path = Path(f.name)
+
+        try:
+            result = validate_video_output(path)
+            assert result.valid is False
+            assert "无视频流" in result.error_message
+        finally:
+            path.unlink()
+
+    @patch("video_processing.video_validation.subprocess.run")
+    def test_ffprobe_timeout(self, mock_run):
+        """ffprobe 超时 → valid=False."""
+        mock_run.side_effect = subprocess.TimeoutExpired(cmd="ffprobe", timeout=15)
+
+        with tempfile.NamedTemporaryFile(suffix=".mp4", delete=False) as f:
+            f.write(b"\x00" * 100)
+            f.write(b"moov")
+            f.write(b"\x00" * 5000)
+            path = Path(f.name)
+
+        try:
+            result = validate_video_output(path)
+            assert result.valid is False
+            assert "超时" in result.error_message
+        finally:
+            path.unlink()
+
+    @patch("video_processing.video_validation.subprocess.run")
+    def test_ffprobe_error(self, mock_run):
+        """ffprobe 执行失败 → valid=False."""
+        mock_run.side_effect = subprocess.CalledProcessError(returncode=1, cmd="ffprobe", stderr="Invalid data found")
+
+        with tempfile.NamedTemporaryFile(suffix=".mp4", delete=False) as f:
+            f.write(b"\x00" * 100)
+            f.write(b"moov")
+            f.write(b"\x00" * 5000)
+            path = Path(f.name)
+
+        try:
+            result = validate_video_output(path)
+            assert result.valid is False
+            assert "ffprobe" in result.error_message.lower()
+        finally:
+            path.unlink()
+
+    @patch("video_processing.video_validation.subprocess.run")
+    def test_short_video_warning(self, mock_run):
+        """极短视频（< min_duration）→ 仍 valid=True 但有警告."""
+        mock_result = MagicMock()
+        mock_result.stdout = json.dumps(
+            {
+                "streams": [{"width": 100, "height": 100, "duration": "0.05", "codec_name": "h264"}],
+                "format": {"duration": "0.05"},
+            }
+        )
+        mock_result.returncode = 0
+        mock_run.return_value = mock_result
+
+        with tempfile.NamedTemporaryFile(suffix=".mp4", delete=False) as f:
+            f.write(b"\x00" * 100)
+            f.write(b"moov")
+            f.write(b"\x00" * 5000)
+            path = Path(f.name)
+
+        try:
+            # 默认 min_duration=0.1，视频只有 0.05s → 应该仍然 valid（仅警告）
+            result = validate_video_output(path)
+            assert result.valid is True
+        finally:
+            path.unlink()
+
+
+# ── VideoValidationResult 数据结构测试 ────────────────────────────────────────
+
+
+class TestVideoValidationResult:
+    """VideoValidationResult 数据结构测试."""
+
+    def test_default_invalid(self):
+        """默认构造结果为无效."""
+        result = VideoValidationResult(valid=False)
+        assert result.valid is False
+        assert result.is_valid is False
+        assert result.file_size == 0
+        assert result.error_message == ""
+
+    def test_valid_result(self):
+        """有效结果属性正确."""
+        result = VideoValidationResult(
+            valid=True,
+            file_exists=True,
+            file_size=1024000,
+            moov_atom_found=True,
+            has_video_stream=True,
+            duration=15.3,
+            width=1920,
+            height=1080,
+        )
+        assert result.is_valid is True
+        assert result.duration == 15.3
+        assert result.width == 1920