diff --git a/apps/worker/video_processing/concat_engine.py b/apps/worker/video_processing/concat_engine.py
index ecebfa60d..166ccfe33 100755
--- a/apps/worker/video_processing/concat_engine.py
+++ b/apps/worker/video_processing/concat_engine.py
@@ -18,135 +18,21 @@
 from __future__ import annotations
 
 import logging
-from dataclasses import dataclass, field
 from pathlib import Path
 
+from packages.domain.video_concat import (  # noqa: F401  向后兼容导出
+    ALLOWED_VIDEO_EXTENSIONS,
+    CONCAT_DEMUXER_REQUIRED_PARAMS,
+    MAX_CONCAT_SEGMENTS,
+    ConcatConfig,
+    ConcatSegment,
+)
 from video_processing.ffmpeg_utils import FFMPEG_BIN, probe_duration, probe_video_info, run_ffmpeg
 from video_processing.path_security import PathSecurityError, is_in_allowed_dirs, safe_resolve_path
 
 logger = logging.getLogger(__name__)
 
 
-# ── 常量 ──────────────────────────────────────────────────────────────────────
-
-MAX_CONCAT_SEGMENTS = 50  # 最大拼接段数（安全上限，防止OOM）
-
-ALLOWED_VIDEO_EXTENSIONS = {".mp4", ".mov", ".avi", ".mkv", ".webm", ".flv", ".wmv"}
-
-# concat demuxer 要求一致的参数列表
-CONCAT_DEMUXER_REQUIRED_PARAMS = [
-    "codec_name",  # 视频编码
-    "width",  # 宽度
-    "height",  # 高度
-    "r_frame_rate",  # 帧率
-    "pix_fmt",  # 像素格式
-    "sample_rate",  # 音频采样率
-    "channels",  # 音频声道数
-    "audio_codec",  # 音频编码
-]
-
-
-# ── 拼接片段配置 ──────────────────────────────────────────────────────────────
-
-
-@dataclass
-class ConcatSegment:
-    """单个拼接片段."""
-
-    video_path: str  # 视频文件路径
-    start_time: float = 0.0  # 开始时间（秒），从视频的哪个位置开始取
-    duration: float = 0.0  # 持续时长（秒），0表示取到末尾
-    has_audio: bool = True  # 是否包含音频
-
-    @classmethod
-    def from_dict(cls, seg: dict) -> "ConcatSegment":
-        """从字典创建拼接片段，带安全类型转换."""
-        try:
-            start_time = max(0.0, float(seg.get("start_time", 0.0)))
-        except (TypeError, ValueError):
-            start_time = 0.0
-
-        try:
-            duration = max(0.0, float(seg.get("duration", 0.0)))
-        except (TypeError, ValueError):
-            duration = 0.0
-
-        return cls(
-            video_path=str(seg.get("video_path", "")),
-            start_time=start_time,
-            duration=duration,
-            has_audio=bool(seg.get("has_audio", True)),
-        )
-
-
-@dataclass
-class ConcatConfig:
-    """视频拼接配置."""
-
-    segments: list[ConcatSegment] = field(default_factory=list)
-    output_width: int = 0  # 输出宽度（0=自动取第一段）
-    output_height: int = 0  # 输出高度（0=自动取第一段）
-    output_fps: float = 0.0  # 输出帧率（0=自动取第一段）
-    force_reencode: bool = False  # 强制重新编码（不用 stream copy）
-    transition: str = "none"  # 转场效果（none/crossfade）- 预留
-    transition_duration: float = 0.3  # 转场时长
-
-    @classmethod
-    def from_config_dict(cls, config: dict | None) -> "ConcatConfig":
-        """从配置字典创建 ConcatConfig."""
-        if not config or not isinstance(config, dict):
-            return cls()
-
-        segments_raw = config.get("segments", [])
-        segments: list[ConcatSegment] = []
-
-        if isinstance(segments_raw, list):
-            for s in segments_raw:
-                if isinstance(s, dict) and s.get("video_path"):
-                    try:
-                        seg = ConcatSegment.from_dict(s)
-                        if seg.video_path:
-                            segments.append(seg)
-                    except Exception:
-                        logger.warning("[concat] skip invalid segment: %s", s)
-                        continue
-
-        try:
-            output_width = max(0, int(config.get("output_width", 0)))
-        except (TypeError, ValueError):
-            output_width = 0
-
-        try:
-            output_height = max(0, int(config.get("output_height", 0)))
-        except (TypeError, ValueError):
-            output_height = 0
-
-        try:
-            output_fps = max(0.0, float(config.get("output_fps", 0.0)))
-        except (TypeError, ValueError):
-            output_fps = 0.0
-
-        return cls(
-            segments=segments,
-            output_width=output_width,
-            output_height=output_height,
-            output_fps=output_fps,
-            force_reencode=bool(config.get("force_reencode", False)),
-            transition=str(config.get("transition", "none")),
-            transition_duration=max(0.1, float(config.get("transition_duration", 0.3))),
-        )
-
-    @property
-    def has_effect(self) -> bool:
-        """是否有有效片段需要拼接."""
-        return len([s for s in self.segments if s.video_path]) >= 2
-
-    @property
-    def total_segments(self) -> int:
-        """有效片段数量."""
-        return len([s for s in self.segments if s.video_path])
-
-
 # ── 路径安全校验 ────────────────────────────────────────────────────────────
 
 
diff --git a/packages/domain/video_concat.py b/packages/domain/video_concat.py
new file mode 100755
index 000000000..da883d207
--- /dev/null
+++ b/packages/domain/video_concat.py
@@ -0,0 +1,177 @@
+"""视频拼接领域模型 — 纯逻辑，无外部依赖.
+
+抽离自 concat_engine.py 的数据类和配置解析逻辑，
+方便单测覆盖，同时保持向后兼容。
+"""
+
+from __future__ import annotations
+
+import logging
+from dataclasses import dataclass, field
+from typing import Any
+
+logger = logging.getLogger(__name__)
+
+# ── 常量 ──────────────────────────────────────────────────────────────────────
+
+MAX_CONCAT_SEGMENTS = 50  # 最大拼接段数（安全上限，防止OOM）
+
+ALLOWED_VIDEO_EXTENSIONS = {".mp4", ".mov", ".avi", ".mkv", ".webm", ".flv", ".wmv"}
+
+# concat demuxer 要求一致的参数列表
+CONCAT_DEMUXER_REQUIRED_PARAMS = [
+    "codec_name",
+    "width",
+    "height",
+    "r_frame_rate",
+    "pix_fmt",
+    "sample_rate",
+    "channels",
+    "audio_codec",
+]
+
+
+# ── 拼接片段配置 ──────────────────────────────────────────────────────────────
+
+
+@dataclass
+class ConcatSegment:
+    """单个拼接片段."""
+
+    video_path: str  # 视频文件路径
+    start_time: float = 0.0  # 开始时间（秒）
+    duration: float = 0.0  # 持续时长（秒），0表示取到末尾
+    has_audio: bool = True  # 是否包含音频
+
+    @classmethod
+    def from_dict(cls, seg: dict[str, Any] | None) -> "ConcatSegment":
+        """从字典创建拼接片段，带安全类型转换."""
+        if not seg or not isinstance(seg, dict):
+            return cls(video_path="")
+
+        try:
+            start_time = max(0.0, float(seg.get("start_time", 0.0)))
+        except (TypeError, ValueError):
+            start_time = 0.0
+
+        try:
+            duration = max(0.0, float(seg.get("duration", 0.0)))
+        except (TypeError, ValueError):
+            duration = 0.0
+
+        return cls(
+            video_path=str(seg.get("video_path", "")),
+            start_time=start_time,
+            duration=duration,
+            has_audio=bool(seg.get("has_audio", True)),
+        )
+
+    @property
+    def is_valid(self) -> bool:
+        """是否为有效片段（有视频路径）."""
+        return bool(self.video_path)
+
+    @property
+    def effective_duration(self) -> float:
+        """有效时长（duration > 0 时取 duration，否则 0）."""
+        return max(0.0, self.duration)
+
+
+@dataclass
+class ConcatConfig:
+    """视频拼接配置."""
+
+    segments: list[ConcatSegment] = field(default_factory=list)
+    output_width: int = 0  # 输出宽度（0=自动取第一段）
+    output_height: int = 0  # 输出高度（0=自动取第一段）
+    output_fps: float = 0.0  # 输出帧率（0=自动取第一段）
+    force_reencode: bool = False  # 强制重新编码
+    transition: str = "none"  # 转场效果（none/crossfade）
+    transition_duration: float = 0.3  # 转场时长
+
+    @classmethod
+    def from_config_dict(cls, config: dict[str, Any] | None) -> "ConcatConfig":
+        """从配置字典创建 ConcatConfig."""
+        if not config or not isinstance(config, dict):
+            return cls()
+
+        segments_raw = config.get("segments", [])
+        segments: list[ConcatSegment] = []
+
+        if isinstance(segments_raw, list):
+            for s in segments_raw:
+                if isinstance(s, dict) and s.get("video_path"):
+                    try:
+                        seg = ConcatSegment.from_dict(s)
+                        if seg.is_valid:
+                            segments.append(seg)
+                    except Exception:
+                        logger.warning("[concat] skip invalid segment: %s", s)
+                        continue
+
+        try:
+            output_width = max(0, int(config.get("output_width", 0)))
+        except (TypeError, ValueError):
+            output_width = 0
+
+        try:
+            output_height = max(0, int(config.get("output_height", 0)))
+        except (TypeError, ValueError):
+            output_height = 0
+
+        try:
+            output_fps = max(0.0, float(config.get("output_fps", 0.0)))
+        except (TypeError, ValueError):
+            output_fps = 0.0
+
+        try:
+            transition_duration = max(0.1, float(config.get("transition_duration", 0.3)))
+        except (TypeError, ValueError):
+            transition_duration = 0.3
+
+        return cls(
+            segments=segments,
+            output_width=output_width,
+            output_height=output_height,
+            output_fps=output_fps,
+            force_reencode=bool(config.get("force_reencode", False)),
+            transition=str(config.get("transition", "none")),
+            transition_duration=transition_duration,
+        )
+
+    @property
+    def has_effect(self) -> bool:
+        """是否有有效片段需要拼接（至少2段）."""
+        return self.valid_segment_count >= 2
+
+    @property
+    def valid_segment_count(self) -> int:
+        """有效片段数量."""
+        return sum(1 for s in self.segments if s.is_valid)
+
+    @property
+    def total_segments(self) -> int:
+        """有效片段数量（向后兼容别名）."""
+        return self.valid_segment_count
+
+    @property
+    def first_valid_segment(self) -> ConcatSegment | None:
+        """第一个有效片段."""
+        for s in self.segments:
+            if s.is_valid:
+                return s
+        return None
+
+    @property
+    def estimated_total_duration(self) -> float:
+        """估算总时长（只统计有明确duration的片段）."""
+        total = 0.0
+        for s in self.segments:
+            if s.is_valid and s.duration > 0:
+                total += s.duration
+        return total
+
+    def clamp_segments(self, max_segments: int = MAX_CONCAT_SEGMENTS) -> None:
+        """截断片段数量，防止OOM."""
+        if len(self.segments) > max_segments:
+            self.segments = self.segments[:max_segments]
diff --git a/tests/unit/test_video_concat.py b/tests/unit/test_video_concat.py
new file mode 100755
index 000000000..7d78551d7
--- /dev/null
+++ b/tests/unit/test_video_concat.py
@@ -0,0 +1,364 @@
+"""video_concat 领域模型单测 — 纯逻辑，48个测试用例."""
+
+from __future__ import annotations
+
+import pytest
+
+from packages.domain.video_concat import (
+    ALLOWED_VIDEO_EXTENSIONS,
+    CONCAT_DEMUXER_REQUIRED_PARAMS,
+    MAX_CONCAT_SEGMENTS,
+    ConcatConfig,
+    ConcatSegment,
+)
+
+# ── ConcatSegment 测试 ───────────────────────────────────────────────────────
+
+
+class TestConcatSegmentBasics:
+    def test_default_values(self):
+        seg = ConcatSegment(video_path="test.mp4")
+        assert seg.video_path == "test.mp4"
+        assert seg.start_time == 0.0
+        assert seg.duration == 0.0
+        assert seg.has_audio is True
+
+    def test_full_params(self):
+        seg = ConcatSegment(
+            video_path="video.mp4",
+            start_time=5.5,
+            duration=10.0,
+            has_audio=False,
+        )
+        assert seg.video_path == "video.mp4"
+        assert seg.start_time == 5.5
+        assert seg.duration == 10.0
+        assert seg.has_audio is False
+
+
+class TestConcatSegmentFromDict:
+    def test_normal_dict(self):
+        seg = ConcatSegment.from_dict(
+            {
+                "video_path": "test.mp4",
+                "start_time": 2.0,
+                "duration": 5.0,
+                "has_audio": False,
+            }
+        )
+        assert seg.video_path == "test.mp4"
+        assert seg.start_time == 2.0
+        assert seg.duration == 5.0
+        assert seg.has_audio is False
+
+    def test_empty_dict(self):
+        seg = ConcatSegment.from_dict({})
+        assert seg.video_path == ""
+        assert seg.start_time == 0.0
+        assert seg.duration == 0.0
+        assert seg.has_audio is True
+
+    def test_none_input(self):
+        seg = ConcatSegment.from_dict(None)
+        assert seg.video_path == ""
+        assert seg.is_valid is False
+
+    def test_non_dict_input(self):
+        seg = ConcatSegment.from_dict("not a dict")
+        assert seg.video_path == ""
+
+    def test_start_time_negative_clamped(self):
+        seg = ConcatSegment.from_dict({"video_path": "a.mp4", "start_time": -5})
+        assert seg.start_time == 0.0
+
+    def test_duration_negative_clamped(self):
+        seg = ConcatSegment.from_dict({"video_path": "a.mp4", "duration": -10})
+        assert seg.duration == 0.0
+
+    def test_start_time_invalid_string(self):
+        seg = ConcatSegment.from_dict({"video_path": "a.mp4", "start_time": "abc"})
+        assert seg.start_time == 0.0
+
+    def test_duration_invalid_string(self):
+        seg = ConcatSegment.from_dict({"video_path": "a.mp4", "duration": "xyz"})
+        assert seg.duration == 0.0
+
+    def test_start_time_int_casted(self):
+        seg = ConcatSegment.from_dict({"video_path": "a.mp4", "start_time": 3})
+        assert seg.start_time == 3.0
+
+    def test_duration_int_casted(self):
+        seg = ConcatSegment.from_dict({"video_path": "a.mp4", "duration": 7})
+        assert seg.duration == 7.0
+
+    def test_video_path_casted_to_string(self):
+        seg = ConcatSegment.from_dict({"video_path": 12345})
+        assert seg.video_path == "12345"
+
+    def test_has_audio_false(self):
+        seg = ConcatSegment.from_dict({"video_path": "a.mp4", "has_audio": False})
+        assert seg.has_audio is False
+
+    def test_has_audio_truthy_value(self):
+        seg = ConcatSegment.from_dict({"video_path": "a.mp4", "has_audio": 1})
+        assert seg.has_audio is True
+
+
+class TestConcatSegmentProperties:
+    def test_is_valid_with_path(self):
+        seg = ConcatSegment(video_path="test.mp4")
+        assert seg.is_valid is True
+
+    def test_is_valid_empty_path(self):
+        seg = ConcatSegment(video_path="")
+        assert seg.is_valid is False
+
+    def test_effective_duration_positive(self):
+        seg = ConcatSegment(video_path="a.mp4", duration=10.5)
+        assert seg.effective_duration == 10.5
+
+    def test_effective_duration_zero(self):
+        seg = ConcatSegment(video_path="a.mp4", duration=0.0)
+        assert seg.effective_duration == 0.0
+
+    def test_effective_duration_negative(self):
+        seg = ConcatSegment(video_path="a.mp4", duration=-5.0)
+        assert seg.effective_duration == 0.0
+
+
+# ── ConcatConfig 测试 ────────────────────────────────────────────────────────
+
+
+class TestConcatConfigBasics:
+    def test_default_values(self):
+        cfg = ConcatConfig()
+        assert cfg.segments == []
+        assert cfg.output_width == 0
+        assert cfg.output_height == 0
+        assert cfg.output_fps == 0.0
+        assert cfg.force_reencode is False
+        assert cfg.transition == "none"
+        assert cfg.transition_duration == 0.3
+
+    def test_with_segments(self):
+        segs = [ConcatSegment(video_path="a.mp4")]
+        cfg = ConcatConfig(segments=segs)
+        assert len(cfg.segments) == 1
+        assert cfg.segments[0].video_path == "a.mp4"
+
+
+class TestConcatConfigFromDict:
+    def test_none_config(self):
+        cfg = ConcatConfig.from_config_dict(None)
+        assert cfg.segments == []
+        assert cfg.output_width == 0
+
+    def test_empty_dict(self):
+        cfg = ConcatConfig.from_config_dict({})
+        assert cfg.segments == []
+
+    def test_non_dict_input(self):
+        cfg = ConcatConfig.from_config_dict("config")
+        assert cfg.segments == []
+
+    def test_with_valid_segments(self):
+        cfg = ConcatConfig.from_config_dict(
+            {
+                "segments": [
+                    {"video_path": "a.mp4", "duration": 10},
+                    {"video_path": "b.mp4", "duration": 20},
+                ],
+            }
+        )
+        assert len(cfg.segments) == 2
+        assert cfg.segments[0].video_path == "a.mp4"
+        assert cfg.segments[1].video_path == "b.mp4"
+
+    def test_skips_empty_video_path(self):
+        cfg = ConcatConfig.from_config_dict(
+            {
+                "segments": [
+                    {"video_path": "a.mp4"},
+                    {"video_path": ""},
+                    {"video_path": "b.mp4"},
+                ],
+            }
+        )
+        assert len(cfg.segments) == 2
+
+    def test_skips_invalid_segment_dict(self):
+        cfg = ConcatConfig.from_config_dict(
+            {
+                "segments": [
+                    {"video_path": "a.mp4"},
+                    "not a dict",
+                    {"video_path": "b.mp4"},
+                ],
+            }
+        )
+        assert len(cfg.segments) == 2
+
+    def test_segments_not_a_list(self):
+        cfg = ConcatConfig.from_config_dict({"segments": "not a list"})
+        assert cfg.segments == []
+
+    def test_output_params(self):
+        cfg = ConcatConfig.from_config_dict(
+            {
+                "output_width": 1920,
+                "output_height": 1080,
+                "output_fps": 30.0,
+                "force_reencode": True,
+            }
+        )
+        assert cfg.output_width == 1920
+        assert cfg.output_height == 1080
+        assert cfg.output_fps == 30.0
+        assert cfg.force_reencode is True
+
+    def test_output_width_negative_clamped(self):
+        cfg = ConcatConfig.from_config_dict({"output_width": -100})
+        assert cfg.output_width == 0
+
+    def test_output_height_invalid_string(self):
+        cfg = ConcatConfig.from_config_dict({"output_height": "abc"})
+        assert cfg.output_height == 0
+
+    def test_output_fps_invalid_string(self):
+        cfg = ConcatConfig.from_config_dict({"output_fps": "xyz"})
+        assert cfg.output_fps == 0.0
+
+    def test_transition_params(self):
+        cfg = ConcatConfig.from_config_dict(
+            {
+                "transition": "crossfade",
+                "transition_duration": 1.0,
+            }
+        )
+        assert cfg.transition == "crossfade"
+        assert cfg.transition_duration == 1.0
+
+    def test_transition_duration_minimum(self):
+        cfg = ConcatConfig.from_config_dict({"transition_duration": 0.01})
+        assert cfg.transition_duration == 0.1
+
+    def test_transition_duration_negative(self):
+        cfg = ConcatConfig.from_config_dict({"transition_duration": -1})
+        assert cfg.transition_duration == 0.1
+
+    def test_force_reencode_false_by_default(self):
+        cfg = ConcatConfig.from_config_dict({})
+        assert cfg.force_reencode is False
+
+
+class TestConcatConfigProperties:
+    def test_has_effect_two_segments(self):
+        cfg = ConcatConfig(
+            segments=[
+                ConcatSegment(video_path="a.mp4"),
+                ConcatSegment(video_path="b.mp4"),
+            ]
+        )
+        assert cfg.has_effect is True
+
+    def test_has_effect_one_segment(self):
+        cfg = ConcatConfig(segments=[ConcatSegment(video_path="a.mp4")])
+        assert cfg.has_effect is False
+
+    def test_has_effect_empty(self):
+        cfg = ConcatConfig()
+        assert cfg.has_effect is False
+
+    def test_has_effect_skips_invalid(self):
+        cfg = ConcatConfig(
+            segments=[
+                ConcatSegment(video_path="a.mp4"),
+                ConcatSegment(video_path=""),
+                ConcatSegment(video_path="b.mp4"),
+            ]
+        )
+        assert cfg.has_effect is True
+
+    def test_valid_segment_count(self):
+        cfg = ConcatConfig(
+            segments=[
+                ConcatSegment(video_path="a.mp4"),
+                ConcatSegment(video_path=""),
+                ConcatSegment(video_path="b.mp4"),
+            ]
+        )
+        assert cfg.valid_segment_count == 2
+
+    def test_first_valid_segment(self):
+        cfg = ConcatConfig(
+            segments=[
+                ConcatSegment(video_path=""),
+                ConcatSegment(video_path="first.mp4"),
+                ConcatSegment(video_path="second.mp4"),
+            ]
+        )
+        assert cfg.first_valid_segment is not None
+        assert cfg.first_valid_segment.video_path == "first.mp4"
+
+    def test_first_valid_segment_none_when_all_empty(self):
+        cfg = ConcatConfig(
+            segments=[
+                ConcatSegment(video_path=""),
+                ConcatSegment(video_path=""),
+            ]
+        )
+        assert cfg.first_valid_segment is None
+
+    def test_first_valid_segment_empty_list(self):
+        cfg = ConcatConfig()
+        assert cfg.first_valid_segment is None
+
+    def test_estimated_total_duration(self):
+        cfg = ConcatConfig(
+            segments=[
+                ConcatSegment(video_path="a.mp4", duration=10.0),
+                ConcatSegment(video_path="b.mp4", duration=20.0),
+                ConcatSegment(video_path="c.mp4", duration=0.0),
+            ]
+        )
+        assert cfg.estimated_total_duration == 30.0
+
+    def test_estimated_total_duration_skips_invalid(self):
+        cfg = ConcatConfig(
+            segments=[
+                ConcatSegment(video_path="", duration=10.0),
+                ConcatSegment(video_path="a.mp4", duration=5.0),
+            ]
+        )
+        assert cfg.estimated_total_duration == 5.0
+
+
+class TestConcatConfigClampSegments:
+    def test_clamp_when_over_max(self):
+        segs = [ConcatSegment(video_path=f"s{i}.mp4") for i in range(100)]
+        cfg = ConcatConfig(segments=segs)
+        cfg.clamp_segments(50)
+        assert len(cfg.segments) == 50
+        assert cfg.segments[0].video_path == "s0.mp4"
+        assert cfg.segments[-1].video_path == "s49.mp4"
+
+    def test_no_clamp_when_under_max(self):
+        segs = [ConcatSegment(video_path=f"s{i}.mp4") for i in range(10)]
+        cfg = ConcatConfig(segments=segs)
+        cfg.clamp_segments(50)
+        assert len(cfg.segments) == 10
+
+    def test_default_max_constant(self):
+        assert MAX_CONCAT_SEGMENTS == 50
+
+
+class TestConstants:
+    def test_allowed_extensions(self):
+        assert ".mp4" in ALLOWED_VIDEO_EXTENSIONS
+        assert ".mov" in ALLOWED_VIDEO_EXTENSIONS
+        assert ".webm" in ALLOWED_VIDEO_EXTENSIONS
+
+    def test_demuxer_params(self):
+        assert "codec_name" in CONCAT_DEMUXER_REQUIRED_PARAMS
+        assert "width" in CONCAT_DEMUXER_REQUIRED_PARAMS
+        assert "r_frame_rate" in CONCAT_DEMUXER_REQUIRED_PARAMS