diff --git a/apps/api/app/api/routes/generation_preview.py b/apps/api/app/api/routes/generation_preview.py
index 5e9414a70..f27d26d6e 100755
--- a/apps/api/app/api/routes/generation_preview.py
+++ b/apps/api/app/api/routes/generation_preview.py
@@ -8,6 +8,7 @@ from __future__ import annotations
 import logging
 
 from app.auth import AuthenticatedUser, get_current_user
+from app.core.storage import get_storage_service
 from app.core.task_enqueue import (
     GLOBAL_PENDING_LIMIT,
     USER_PENDING_LIMIT,
@@ -63,6 +64,29 @@ def _mark_task_failed(repo, task, reason: str) -> None:
         logger.exception("[预览生成] 标记任务失败时异常: task_id=%s", task.id)
 
 
+def _sign_video_url(raw_url: str) -> str:
+    """为私有 OSS bucket 的视频 URL 生成预签名下载链接。
+
+    有效期 2 小时，签名失败时降级返回原始 URL。
+    """
+    if not raw_url:
+        return ""
+    try:
+        storage = get_storage_service()
+        signed = storage.get_download_url(raw_url, expires_seconds=7200)
+        # 如果返回的 URL 与原始 URL 完全不同且不是签名 URL（说明 bucket 未配置），
+        # 降级返回原始 URL
+        if signed and signed != raw_url:
+            return signed
+        if signed == raw_url:
+            return raw_url
+        # signed 为空或与 raw_url 无关，返回原始
+        return raw_url
+    except Exception:
+        logger.warning("[预览] URL签名失败，降级返回原始URL: %s", raw_url[:100], exc_info=True)
+        return raw_url
+
+
 def _to_preview_response(task, generated_videos: list | None = None) -> PreviewGenerationTaskResponse:
     """将领域任务对象转换为预览响应 DTO。
 
@@ -78,7 +102,9 @@ def _to_preview_response(task, generated_videos: list | None = None) -> PreviewG
     file_size = 0
     if generated_videos:
         first_video = generated_videos[0]
-        video_url = getattr(first_video, "file_url", "") or ""
+        raw_url = getattr(first_video, "file_url", "") or ""
+        # P0 修复：私有 bucket 需要预签名 URL，否则前端 403 → 黑屏
+        video_url = _sign_video_url(raw_url)
         duration = float(getattr(first_video, "duration", 0.0) or 0.0)
         file_size = int(getattr(first_video, "file_size", 0) or 0)
 
diff --git a/apps/worker/worker_app/tasks/generation.py b/apps/worker/worker_app/tasks/generation.py
index 6d1deebd5..fb917d7d1 100644
--- a/apps/worker/worker_app/tasks/generation.py
+++ b/apps/worker/worker_app/tasks/generation.py
@@ -189,6 +189,43 @@ def _load_template_clip_configs(template_id: str) -> list:
         return []
 
 
+def _load_template_segment_durations(template_id: str) -> list[float]:
+    """从数据库读取模板各 segment 的 duration_max 列表（按 segment_order 排序）。
+
+    用于限制每个 clip 的最大时长，防止素材完整时长超过模板约束。
+    失败返回空列表，不阻断主流程。
+    """
+    if not template_id:
+        return []
+    try:
+        from worker_app.db import SessionLocal
+
+        from packages.adapters.sqlalchemy_impl.models import TemplateSegmentModel
+
+        session = SessionLocal()
+        try:
+            segments = (
+                session.query(TemplateSegmentModel)
+                .filter(TemplateSegmentModel.template_id == template_id)
+                .order_by(TemplateSegmentModel.segment_order)
+                .all()
+            )
+            durations = [s.duration_max for s in segments if s.duration_max and s.duration_max > 0]
+            if durations:
+                logger.info(
+                    "读取模板segment时长约束: template_id=%s segments=%d durations=%s",
+                    template_id,
+                    len(durations),
+                    durations,
+                )
+            return durations
+        finally:
+            session.close()
+    except Exception as e:
+        logger.warning("读取模板segment时长约束失败: template_id=%s error=%s", template_id, e)
+        return []
+
+
 def _build_plan_and_clips_from_task(
     task_id: str,
     downloaded_paths: list[Path],
@@ -281,6 +318,25 @@ def _build_plan_and_clips_from_task(
                 )
             )
 
+    # ── P1-2: 模板 segment 时长约束 ──
+    if template_id and clips:
+        seg_durations = _load_template_segment_durations(template_id)
+        if seg_durations:
+            capped_count = 0
+            for idx, clip in enumerate(clips):
+                if idx < len(seg_durations):
+                    max_dur = seg_durations[idx]
+                    if clip.duration > max_dur:
+                        clip.duration = max_dur
+                        capped_count += 1
+            if capped_count > 0:
+                logger.info(
+                    "模板时长约束已应用: template_id=%s capped_clips=%d/%d",
+                    template_id,
+                    capped_count,
+                    len(clips),
+                )
+
     # ── P1: 模板效果层映射 ──
     if template_id:
         clip_configs = _load_template_clip_configs(template_id)
@@ -1100,13 +1156,14 @@ def _render_video(
     plan_cfg = dict(virtual_plan.config or {})
     export_cfg = dict(plan_cfg.get("export", {}) or {})
     if is_preview:
-        # 预览模式强制 480p + 低码率
-        export_cfg["resolution"] = "854x480"
+        # 预览模式：短边 480p + 低码率，但尊重视频比例（竖屏模板不应强制横屏）
+        preview_res = resolution if resolution else "854x480"
+        export_cfg["resolution"] = preview_res
         export_cfg["bitrate"] = "1M"
         logger.info(
-            "[task_id=%s] [渲染] 预览模式：强制分辨率=%s, 码率=%s",
+            "[task_id=%s] [渲染] 预览模式：分辨率=%s, 码率=%s",
             task_id,
-            "854x480",
+            preview_res,
             "1M",
         )
     elif resolution:
diff --git a/tests/unit/test_generation_preview.py b/tests/unit/test_generation_preview.py
index a2f5a0936..de1ad7bec 100755
--- a/tests/unit/test_generation_preview.py
+++ b/tests/unit/test_generation_preview.py
@@ -631,7 +631,7 @@ class TestToPreviewResponse:
         assert resp.file_size == 0
 
     def test_completed_task_with_videos(self):
-        """已完成任务，带视频结果"""
+        """已完成任务，带视频结果（URL签名后返回）"""
         task = _make_task(
             status=GenerationTaskStatus.COMPLETED,
             progress=100.0,
@@ -640,8 +640,12 @@ class TestToPreviewResponse:
         video.file_url = "https://cdn.example.com/preview.mp4"
         video.duration = 30.5
         video.file_size = 1024000
-        resp = _to_preview_response(task, generated_videos=[video])
-        assert resp.video_url == "https://cdn.example.com/preview.mp4"
+        # Mock storage service to return a signed URL
+        mock_storage = MagicMock()
+        mock_storage.get_download_url.return_value = "https://cdn.example.com/preview.mp4?sig=test123"
+        with patch("app.api.routes.generation_preview.get_storage_service", return_value=mock_storage):
+            resp = _to_preview_response(task, generated_videos=[video])
+        assert resp.video_url == "https://cdn.example.com/preview.mp4?sig=test123"
         assert resp.duration == 30.5
         assert resp.file_size == 1024000
 
@@ -963,7 +967,7 @@ class TestGetPreviewRoute:
             assert exc_info.value.status_code == 403
 
     def test_get_completed_task_with_videos(self):
-        """查询 completed 状态任务，返回视频列表"""
+        """查询 completed 状态任务，返回视频列表（URL 已签名）"""
         repo = MagicMock()
         vid_repo = MagicMock()
         task = _make_task(status=GenerationTaskStatus.COMPLETED, progress=100.0)
@@ -973,17 +977,22 @@ class TestGetPreviewRoute:
         video.duration = 25.0
         video.file_size = 512000
 
+        # Mock URL 签名（返回带签名的 URL）
+        mock_storage = MagicMock()
+        mock_storage.get_download_url.return_value = "https://cdn.example.com/preview_final.mp4?sig=abc123"
+
         with patch("app.api.routes.generation_preview.GetGenerationTaskUseCase") as MockGet:
             MockGet.return_value.execute.return_value = task
             with patch("app.api.routes.generation_preview.ListGeneratedVideosByTaskUseCase") as MockList:
                 MockList.return_value.execute.return_value = [video]
-                resp = get_preview_generation_task(
-                    task_id="preview_task_001",
-                    authenticated_user=_make_user(),
-                    generation_task_repository=repo,
-                    generated_video_repository=vid_repo,
-                )
-        assert resp.video_url == "https://cdn.example.com/preview_final.mp4"
+                with patch("app.api.routes.generation_preview.get_storage_service", return_value=mock_storage):
+                    resp = get_preview_generation_task(
+                        task_id="preview_task_001",
+                        authenticated_user=_make_user(),
+                        generation_task_repository=repo,
+                        generated_video_repository=vid_repo,
+                    )
+        assert resp.video_url == "https://cdn.example.com/preview_final.mp4?sig=abc123"
         assert resp.duration == 25.0
 
 
@@ -995,37 +1004,38 @@ class TestGetPreviewRoute:
 class TestWorkerPreviewResolution:
     """Worker 层预览模式分辨率强制逻辑测试"""
 
-    def test_preview_mode_forces_480p(self):
-        """预览模式强制 854x480 + 1M 码率"""
-        # 模拟 worker 中 export_cfg 逻辑（与实际代码一致，使用 dict 拷贝）
+    def test_preview_mode_respects_resolution(self):
+        """预览模式尊重传入的 resolution 参数（如竖屏 480x854）"""
         is_preview = True
-        resolution = "1920x1080"  # 用户指定的分辨率应被忽略
+        resolution = "480x854"  # 竖屏模板
         original_config = {"export": {"resolution": "1280x720", "bitrate": "5M"}}
         plan_cfg = dict(original_config)
         export_cfg = dict(plan_cfg.get("export", {}) or {})
 
         if is_preview:
-            export_cfg["resolution"] = "854x480"
+            preview_res = resolution if resolution else "854x480"
+            export_cfg["resolution"] = preview_res
             export_cfg["bitrate"] = "1M"
         elif resolution:
             export_cfg["resolution"] = resolution
 
         plan_cfg["export"] = export_cfg
 
-        assert export_cfg["resolution"] == "854x480"
+        assert export_cfg["resolution"] == "480x854"  # 竖屏
         assert export_cfg["bitrate"] == "1M"
         # 验证原始配置未被污染
         assert original_config["export"]["resolution"] == "1280x720"
 
     def test_non_preview_uses_user_resolution(self):
-        """非预览模式使用用户指定分辨率"""
+        """非预览模式使用用户指定分辨率（逻辑不变）"""
         is_preview = False
         resolution = "1920x1080"
         plan_cfg = dict({"export": {"resolution": "1280x720"}})
         export_cfg = dict(plan_cfg.get("export", {}) or {})
 
         if is_preview:
-            export_cfg["resolution"] = "854x480"
+            preview_res = resolution if resolution else "854x480"
+            export_cfg["resolution"] = preview_res
             export_cfg["bitrate"] = "1M"
         elif resolution:
             export_cfg["resolution"] = resolution
@@ -1033,14 +1043,15 @@ class TestWorkerPreviewResolution:
         assert export_cfg["resolution"] == "1920x1080"
 
     def test_non_preview_no_resolution_uses_template(self):
-        """非预览模式且用户未指定分辨率，使用模板配置"""
+        """非预览模式且用户未指定分辨率，使用模板配置（逻辑不变）"""
         is_preview = False
         resolution = ""
         plan_cfg = dict({"export": {"resolution": "1280x720"}})
         export_cfg = dict(plan_cfg.get("export", {}) or {})
 
         if is_preview:
-            export_cfg["resolution"] = "854x480"
+            preview_res = resolution if resolution else "854x480"
+            export_cfg["resolution"] = preview_res
             export_cfg["bitrate"] = "1M"
         elif resolution:
             export_cfg["resolution"] = resolution
@@ -1050,14 +1061,94 @@ class TestWorkerPreviewResolution:
         assert export_cfg["resolution"] == "1280x720"
 
     def test_preview_mode_empty_export_cfg(self):
-        """预览模式且模板无 export 配置"""
+        """预览模式且模板无 export 配置，无传入 resolution 时默认 854x480"""
         is_preview = True
+        resolution = ""
         plan_cfg = dict({})
         export_cfg = dict(plan_cfg.get("export", {}) or {})
 
         if is_preview:
-            export_cfg["resolution"] = "854x480"
+            preview_res = resolution if resolution else "854x480"
+            export_cfg["resolution"] = preview_res
             export_cfg["bitrate"] = "1M"
 
         assert export_cfg["resolution"] == "854x480"
         assert export_cfg["bitrate"] == "1M"
+
+
+# ═══════════════════════════════════════════════════════════════════════════════
+# 新增：URL 签名 + 模板时长约束测试
+# ═══════════════════════════════════════════════════════════════════════════════
+
+
+class TestSignVideoUrl:
+    """_sign_video_url 预签名 URL 测试。"""
+
+    def test_empty_url_returns_empty(self):
+        """空 URL 直接返回空字符串。"""
+        from app.api.routes.generation_preview import _sign_video_url
+
+        assert _sign_video_url("") == ""
+
+    def test_signs_oss_url(self):
+        """OSS URL 应被签名。"""
+        from app.api.routes.generation_preview import _sign_video_url
+
+        mock_storage = MagicMock()
+        mock_storage.get_download_url.return_value = "https://signed.example.com/video.mp4?sig=abc"
+
+        with patch("app.api.routes.generation_preview.get_storage_service", return_value=mock_storage):
+            result = _sign_video_url("https://bucket.oss.example.com/video.mp4")
+
+        assert result == "https://signed.example.com/video.mp4?sig=abc"
+        mock_storage.get_download_url.assert_called_once()
+
+    def test_fallback_on_sign_failure(self):
+        """签名失败时降级返回原始 URL。"""
+        from app.api.routes.generation_preview import _sign_video_url
+
+        with patch("app.api.routes.generation_preview.get_storage_service", side_effect=RuntimeError("no storage")):
+            result = _sign_video_url("https://bucket.oss.example.com/video.mp4")
+
+        assert result == "https://bucket.oss.example.com/video.mp4"
+
+    def test_sign_returns_none_fallback(self):
+        """get_download_url 返回 None 时降级返回原始 URL。"""
+        from app.api.routes.generation_preview import _sign_video_url
+
+        mock_storage = MagicMock()
+        mock_storage.get_download_url.return_value = None
+
+        with patch("app.api.routes.generation_preview.get_storage_service", return_value=mock_storage):
+            result = _sign_video_url("https://bucket.oss.example.com/video.mp4")
+
+        assert result == "https://bucket.oss.example.com/video.mp4"
+
+
+class TestCalcPreviewResolution:
+    """_calc_preview_resolution 单元测试。"""
+
+    def test_portrait_9_16(self):
+        from app.api.routes.generation_preview import _calc_preview_resolution
+
+        assert _calc_preview_resolution("9:16") == "480x854"
+
+    def test_landscape_16_9(self):
+        from app.api.routes.generation_preview import _calc_preview_resolution
+
+        assert _calc_preview_resolution("16:9") == "854x480"
+
+    def test_square_1_1(self):
+        from app.api.routes.generation_preview import _calc_preview_resolution
+
+        assert _calc_preview_resolution("1:1") == "480x480"
+
+    def test_unknown_defaults_to_landscape(self):
+        from app.api.routes.generation_preview import _calc_preview_resolution
+
+        assert _calc_preview_resolution("unknown") == "854x480"
+
+    def test_empty_defaults_to_landscape(self):
+        from app.api.routes.generation_preview import _calc_preview_resolution
+
+        assert _calc_preview_resolution("") == "854x480"
diff --git a/tests/unit/test_generation_worker_fixes.py b/tests/unit/test_generation_worker_fixes.py
new file mode 100644
index 000000000..70c43294c
--- /dev/null
+++ b/tests/unit/test_generation_worker_fixes.py
@@ -0,0 +1,209 @@
+"""Tests for generation.py worker-side fixes: segment durations + preview resolution."""
+
+from __future__ import annotations
+
+import os
+import sys
+from pathlib import Path
+from unittest.mock import MagicMock, patch
+
+# Add worker app to path
+sys.path.insert(0, str(Path(__file__).resolve().parents[2] / "apps" / "worker"))
+os.environ.setdefault("JWT_SECRET_KEY", "unit-test-secret")
+os.environ.setdefault("DATABASE_URL", "sqlite:///test.db")
+
+
+class TestLoadTemplateSegmentDurations:
+    """_load_template_segment_durations 单元测试 (covers lines 198-226)."""
+
+    def test_empty_template_id(self):
+        """空 template_id 直接返回空列表。"""
+        from worker_app.tasks.generation import _load_template_segment_durations
+
+        result = _load_template_segment_durations("")
+        assert result == []
+
+    def test_loads_durations_ordered(self):
+        """按 segment_order 排序返回 duration_max 列表。"""
+        from worker_app.tasks.generation import _load_template_segment_durations
+
+        mock_seg1 = MagicMock(duration_max=5.0)
+        mock_seg2 = MagicMock(duration_max=8.0)
+        mock_seg3 = MagicMock(duration_max=3.0)
+
+        mock_query = MagicMock()
+        mock_query.filter.return_value.order_by.return_value.all.return_value = [
+            mock_seg1,
+            mock_seg2,
+            mock_seg3,
+        ]
+        mock_session = MagicMock()
+        mock_session.query.return_value = mock_query
+
+        # Patch at the source module since it's imported inside the function
+        with patch("worker_app.db.SessionLocal", return_value=mock_session):
+            result = _load_template_segment_durations("tpl_123")
+
+        assert result == [5.0, 8.0, 3.0]
+
+    def test_filters_zero_and_negative(self):
+        """duration_max <= 0 的 segment 被过滤。"""
+        from worker_app.tasks.generation import _load_template_segment_durations
+
+        mock_seg_valid = MagicMock(duration_max=5.0)
+        mock_seg_zero = MagicMock(duration_max=0.0)
+        mock_seg_none = MagicMock(duration_max=None)
+
+        mock_query = MagicMock()
+        mock_query.filter.return_value.order_by.return_value.all.return_value = [
+            mock_seg_valid,
+            mock_seg_zero,
+            mock_seg_none,
+        ]
+        mock_session = MagicMock()
+        mock_session.query.return_value = mock_query
+
+        with patch("worker_app.db.SessionLocal", return_value=mock_session):
+            result = _load_template_segment_durations("tpl_456")
+
+        assert result == [5.0]
+
+    def test_db_error_returns_empty(self):
+        """数据库异常返回空列表，不抛出。"""
+        from worker_app.tasks.generation import _load_template_segment_durations
+
+        with patch("worker_app.db.SessionLocal", side_effect=Exception("DB down")):
+            result = _load_template_segment_durations("tpl_789")
+
+        assert result == []
+
+    def test_empty_segments_returns_empty(self):
+        """没有 segment 时返回空列表。"""
+        from worker_app.tasks.generation import _load_template_segment_durations
+
+        mock_query = MagicMock()
+        mock_query.filter.return_value.order_by.return_value.all.return_value = []
+        mock_session = MagicMock()
+        mock_session.query.return_value = mock_query
+
+        with patch("worker_app.db.SessionLocal", return_value=mock_session):
+            result = _load_template_segment_durations("tpl_empty")
+
+        assert result == []
+
+
+class TestDurationCappingInBuildPlan:
+    """_build_plan_and_clips_from_task 时长约束测试 (covers lines 322-333)."""
+
+    def _make_temp_video(self, tmpdir: Path, name: str = "v.mp4") -> Path:
+        p = tmpdir / name
+        p.write_bytes(b"\x00" * 100)
+        return p
+
+    def test_clips_capped_by_segment_max(self):
+        """clip 时长超过 segment duration_max 时截断。"""
+        import tempfile
+
+        from worker_app.tasks.generation import _build_plan_and_clips_from_task
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            tmpdir = Path(tmpdir)
+            paths = [self._make_temp_video(tmpdir, f"v{i}.mp4") for i in range(3)]
+
+            with patch("worker_app.tasks.generation.probe_duration", return_value=30.0):
+                with patch(
+                    "worker_app.tasks.generation._load_template_segment_durations",
+                    return_value=[5.0, 4.0, 3.0],
+                ):
+                    with patch(
+                        "worker_app.tasks.generation._load_template_clip_configs",
+                        return_value=[],
+                    ):
+                        _, clips, _ = _build_plan_and_clips_from_task(
+                            task_id="test_cap",
+                            downloaded_paths=paths,
+                            mode="one_take",
+                            template_id="tpl_test",
+                        )
+
+            assert clips[0].duration == 5.0
+            assert clips[1].duration == 4.0
+            assert clips[2].duration == 3.0
+
+    def test_clips_not_capped_when_under_max(self):
+        """clip 时长小于 segment duration_max 时不截断。"""
+        import tempfile
+
+        from worker_app.tasks.generation import _build_plan_and_clips_from_task
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            tmpdir = Path(tmpdir)
+            paths = [self._make_temp_video(tmpdir)]
+
+            with patch("worker_app.tasks.generation.probe_duration", return_value=3.0):
+                with patch(
+                    "worker_app.tasks.generation._load_template_segment_durations",
+                    return_value=[5.0],
+                ):
+                    with patch(
+                        "worker_app.tasks.generation._load_template_clip_configs",
+                        return_value=[],
+                    ):
+                        _, clips, _ = _build_plan_and_clips_from_task(
+                            task_id="test_no_cap",
+                            downloaded_paths=paths,
+                            mode="one_take",
+                            template_id="tpl_test",
+                        )
+
+            assert clips[0].duration == 3.0
+
+    def test_no_capping_without_template(self):
+        """无 template_id 时不截断。"""
+        import tempfile
+
+        from worker_app.tasks.generation import _build_plan_and_clips_from_task
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            tmpdir = Path(tmpdir)
+            paths = [self._make_temp_video(tmpdir)]
+
+            with patch("worker_app.tasks.generation.probe_duration", return_value=30.0):
+                _, clips, _ = _build_plan_and_clips_from_task(
+                    task_id="test_no_tpl",
+                    downloaded_paths=paths,
+                    mode="one_take",
+                    template_id="",
+                )
+
+            assert clips[0].duration == 30.0
+
+    def test_partial_segments_only_caps_matching(self):
+        """segment 数量少于 clip 时，只截断有对应 segment 的 clip。"""
+        import tempfile
+
+        from worker_app.tasks.generation import _build_plan_and_clips_from_task
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            tmpdir = Path(tmpdir)
+            paths = [self._make_temp_video(tmpdir, f"v{i}.mp4") for i in range(3)]
+
+            with patch("worker_app.tasks.generation.probe_duration", return_value=20.0):
+                with patch(
+                    "worker_app.tasks.generation._load_template_segment_durations",
+                    return_value=[5.0],  # only 1 segment for 3 clips
+                ):
+                    with patch(
+                        "worker_app.tasks.generation._load_template_clip_configs",
+                        return_value=[],
+                    ):
+                        _, clips, _ = _build_plan_and_clips_from_task(
+                            task_id="test_partial",
+                            downloaded_paths=paths,
+                            mode="one_take",
+                            template_id="tpl_test",
+                        )
+
+            assert clips[0].duration == 5.0  # capped
+            assert clips[1].duration == 20.0  # not capped (no matching segment)
+            assert clips[2].duration == 20.0  # not capped