diff --git a/apps/api/app/core/storage.py b/apps/api/app/core/storage.py
index cf6997b61..566a5d3c7 100644
--- a/apps/api/app/core/storage.py
+++ b/apps/api/app/core/storage.py
@@ -166,12 +166,28 @@ class OSSStorageService:
         if self.bucket is None:
             if self._is_local_generated_url(storage_key_or_url):
                 return storage_key_or_url
+            logger.warning(
+                "get_download_url: OSS bucket not configured, returning raw URL. "
+                "storage_key_or_url=%s",
+                storage_key_or_url[:200],
+            )
             return self.get_url(self._normalize_storage_key(storage_key_or_url))
 
         storage_key = self._normalize_storage_key(storage_key_or_url)
         try:
-            return self.bucket.sign_url("GET", storage_key, expires_seconds)
+            signed = self.bucket.sign_url("GET", storage_key, expires_seconds)
+            logger.info(
+                "get_download_url: signed URL generated. storage_key=%s url_prefix=%s",
+                storage_key[:80],
+                signed[:60],
+            )
+            return signed
         except Exception:
+            logger.exception(
+                "get_download_url: sign_url failed, falling back to raw URL. "
+                "storage_key=%s",
+                storage_key[:200],
+            )
             return self.get_url(storage_key)
 
     def _normalize_storage_key(self, storage_key_or_url: str) -> str:
diff --git a/apps/worker/video_processing/ffmpeg_utils.py b/apps/worker/video_processing/ffmpeg_utils.py
index f1a68377d..79fecb58d 100644
--- a/apps/worker/video_processing/ffmpeg_utils.py
+++ b/apps/worker/video_processing/ffmpeg_utils.py
@@ -250,8 +250,11 @@ def build_xfade_filter_chain(
 ) -> tuple[str, float]:
     """构建 xfade 转场滤镜链。
 
+    对每步 xfade 自动钳制 transition duration，确保
+    ``offset + td ≤ first_input_duration``，避免 FFmpeg exit 234。
+
     Args:
-        clip_durations: 每个片段的时长
+        clip_durations: 每个片段的时长（必须与 trim 后的实际时长一致）
         clip_video_labels: 每个片段的视频流标签（如 "v0", "v1"）
         transitions: 每个片段对应的转场效果（第一个片段的转场被忽略）
         transition_duration: 转场时长（秒）
@@ -262,23 +265,42 @@ def build_xfade_filter_chain(
     """
     n = len(clip_durations)
     parts: list[str] = []
-    total_duration = sum(clip_durations)
 
     if n == 0:
         return "", 0.0
 
     if n == 1:
         parts.append(f"[{clip_video_labels[0]}]copy[{output_label}]")
-        return ";".join(parts), total_duration
+        return ";".join(parts), clip_durations[0]
 
-    # xfade 链
+    # xfade 链 — 每步动态钳制 td，防止 offset + td > first_input_duration
     cumulative = 0.0
     prev_label = clip_video_labels[0]
+    total_transition = 0.0  # 累计已使用的转场时长
 
     for i in range(1, n):
         cumulative += clip_durations[i - 1]
+
+        # 当前 xfade 的第一个输入时长
+        if i == 1:
+            first_input_dur = clip_durations[0]
+        else:
+            first_input_dur = cumulative - total_transition
+
+        # 原始 offset 计算
         offset = max(0.0, cumulative - transition_duration * i)
 
+        # 安全钳制：offset + td 不能超过第一个输入的时长
+        available = max(0.0, first_input_dur - offset)
+        safe_td = min(transition_duration, available)
+
+        # 同时不能超过剩余总时长
+        remaining = max(0.0, sum(clip_durations) - cumulative)
+        safe_td = min(safe_td, remaining)
+        # 同时不能超过当前第二个输入（单个片段）的时长
+        safe_td = min(safe_td, clip_durations[i])
+        safe_td = max(0.001, safe_td)  # 至少 1ms，避免 td=0
+
         transition = transitions[i] if i < len(transitions) else "cut"
         xfade_transition = resolve_xfade_transition(transition)
 
@@ -290,12 +312,13 @@ def build_xfade_filter_chain(
         parts.append(
             f"[{prev_label}][{clip_video_labels[i]}]"
             f"xfade=transition={xfade_transition}"
-            f":duration={transition_duration}"
+            f":duration={safe_td:.3f}"
             f":offset={offset:.3f}"
             f"[{out_label}]"
         )
         prev_label = out_label
+        total_transition += safe_td
 
     # 总时长减去转场重叠部分
-    total_duration -= transition_duration * (n - 1)
+    total_duration = sum(clip_durations) - total_transition
     return ";".join(parts), max(0.0, total_duration)
diff --git a/apps/worker/video_processing/unified_render_service.py b/apps/worker/video_processing/unified_render_service.py
index 8852ceb67..a01c42c7e 100644
--- a/apps/worker/video_processing/unified_render_service.py
+++ b/apps/worker/video_processing/unified_render_service.py
@@ -310,9 +310,16 @@ class UnifiedRenderService:
 
             filters: list[str] = []
 
-            # trim（如果指定了 duration）
-            if clip.duration > 0 and clip.duration < clip.actual_duration:
-                filters.append(f"trim=duration={clip.duration}")
+            # trim — 始终将输出截断到有效时长，防止 xfade offset 与实际时长不匹配
+            # 有效时长 = min(指定时长, 实际时长)；若均未设置则跳过
+            effective_duration = 0.0
+            if clip.duration > 0:
+                effective_duration = min(clip.duration, clip.actual_duration) if clip.actual_duration > 0 else clip.duration
+            elif clip.actual_duration > 0:
+                effective_duration = clip.actual_duration
+
+            if effective_duration > 0:
+                filters.append(f"trim=duration={effective_duration}")
                 filters.append("setpts=PTS-STARTPTS")
 
             # scale
@@ -344,10 +351,15 @@ class UnifiedRenderService:
         for layer in layers:
             layer_clip_indices = [all_clips.index(c) for c in layer.clips]
             layer_labels = [preprocessed_labels[i] for i in layer_clip_indices]
-            layer_durations = [
-                all_clips[i].duration if all_clips[i].duration > 0 else all_clips[i].actual_duration
-                for i in layer_clip_indices
-            ]
+            # 使用 trim 后的有效时长，与 Step 1 的 trim=duration 保持一致
+            layer_durations = []
+            for i in layer_clip_indices:
+                c = all_clips[i]
+                if c.duration > 0:
+                    eff = min(c.duration, c.actual_duration) if c.actual_duration > 0 else c.duration
+                else:
+                    eff = c.actual_duration if c.actual_duration > 0 else 0.0
+                layer_durations.append(eff)
             layer_transitions = [all_clips[i].transition_effect for i in layer_clip_indices]
 
             if len(layer_labels) == 1:
diff --git a/postgres b/postgres
new file mode 100644
index 000000000..e69de29bb
diff --git a/tests/unit/test_p02_p03_fixes.py b/tests/unit/test_p02_p03_fixes.py
new file mode 100644
index 000000000..d24ac4655
--- /dev/null
+++ b/tests/unit/test_p02_p03_fixes.py
@@ -0,0 +1,379 @@
+"""P0-2 / P0-3 修复验证测试.
+
+P0-2: OSSStorageService.get_download_url 预签名 URL 逻辑验证
+P0-3: FFmpeg xfade exit 234 — build_xfade_filter_chain 安全钳制 + effective_duration trim
+"""
+
+from __future__ import annotations
+
+import logging
+from unittest.mock import MagicMock, patch
+
+import pytest
+from video_processing.ffmpeg_utils import build_xfade_filter_chain
+
+# ── P0-3: build_xfade_filter_chain 安全钳制 ──────────────────────────────────
+
+
+
+class TestBuildXfadeFilterChainSafetyClamp:
+    """验证 xfade 滤镜链的安全钳制逻辑，防止 exit 234。"""
+
+    def test_empty_clips(self):
+        """空片段列表返回空字符串和 0 时长。"""
+        f, dur = build_xfade_filter_chain([], [], [])
+        assert f == ""
+        assert dur == 0.0
+
+    def test_single_clip(self):
+        """单片段用 copy，不做 xfade。"""
+        f, dur = build_xfade_filter_chain([3.0], ["v0"], ["cut"])
+        assert "copy" in f
+        assert dur == 3.0
+
+    def test_two_clips_normal(self):
+        """两个正常时长片段 — offset + td ≤ first_input_duration。"""
+        f, dur = build_xfade_filter_chain(
+            clip_durations=[5.0, 5.0],
+            clip_video_labels=["v0", "v1"],
+            transitions=["cut", "fade"],
+            transition_duration=0.5,
+        )
+        assert "xfade" in f
+        assert "duration=0.500" in f
+        # 总时长 = 5 + 5 - 0.5 = 9.5
+        assert abs(dur - 9.5) < 0.01
+
+    def test_short_clip_safety_clamp(self):
+        """片段短于 transition_duration 时，td 被钳制，不会导致 exit 234。
+
+        这是 P0-3 的核心场景：视频只有 1s，td=0.5s，offset 计算后
+        offset + td 可能超过 first_input_duration。
+        """
+        # 两个 1s 片段，td=0.5s
+        # 原始 offset = max(0, 1.0 - 0.5*1) = 0.5
+        # first_input_dur = 1.0, available = 1.0 - 0.5 = 0.5
+        # safe_td = min(0.5, 0.5) = 0.5 — 刚好 OK
+        f, dur = build_xfade_filter_chain(
+            clip_durations=[1.0, 1.0],
+            clip_video_labels=["v0", "v1"],
+            transitions=["cut", "fade"],
+            transition_duration=0.5,
+        )
+        assert "xfade" in f
+        # offset + td 必须 ≤ first_input_duration
+        # offset=0.5, td=0.5 → 0.5+0.5=1.0 ≤ 1.0 ✓
+        assert dur > 0
+
+    def test_very_short_clip_clamped(self):
+        """片段极短（0.3s），td 被钳制到 available 以内。"""
+        # clip1=0.3s, clip2=5.0s, td=0.5s
+        # offset = max(0, 0.3 - 0.5) = 0.0
+        # first_input_dur = 0.3
+        # available = 0.3 - 0.0 = 0.3
+        # safe_td = min(0.5, 0.3) = 0.3
+        f, dur = build_xfade_filter_chain(
+            clip_durations=[0.3, 5.0],
+            clip_video_labels=["v0", "v1"],
+            transitions=["cut", "fade"],
+            transition_duration=0.5,
+        )
+        assert "duration=0.300" in f  # td 被钳制到 0.3
+
+    def test_multi_clip_chain_clamp(self):
+        """多片段链式 xfade，每步都钳制。"""
+        # 3 个 0.5s 片段，td=0.5s
+        # Step 1: offset=0, first_input_dur=0.5, available=0.5, safe_td=0.5
+        #   total_transition=0.5
+        # Step 2: cumulative=1.0, first_input_dur=1.0-0.5=0.5
+        #   offset=max(0, 1.0-0.5*2)=0.0, available=0.5, safe_td=0.5
+        f, dur = build_xfade_filter_chain(
+            clip_durations=[0.5, 0.5, 0.5],
+            clip_video_labels=["v0", "v1", "v2"],
+            transitions=["cut", "fade", "fade"],
+            transition_duration=0.5,
+        )
+        assert "xfade" in f
+        assert f.count("xfade") == 2
+        assert dur > 0
+
+    def test_middle_clip_shorter_than_td(self):
+        """P1 修复验证：中间片段短于 td 时，td 被钳制到该片段时长。
+
+        [5.0, 0.3, 5.0] + td=0.5s:
+        - Step 1: second input = 0.3s, safe_td 必须 ≤ 0.3
+        - Step 2: second input = 5.0s, safe_td 可以 = 0.5
+        """
+        import re
+
+        f, dur = build_xfade_filter_chain(
+            clip_durations=[5.0, 0.3, 5.0],
+            clip_video_labels=["v0", "v1", "v2"],
+            transitions=["cut", "fade", "fade"],
+            transition_duration=0.5,
+        )
+        assert f.count("xfade") == 2
+
+        # 解析每个 xfade 的 duration
+        durations_found = []
+        for part in f.split(";"):
+            if "xfade=" not in part:
+                continue
+            m = re.search(r"duration=([\d.]+)", part)
+            assert m, f"无法解析: {part}"
+            durations_found.append(float(m.group(1)))
+
+        # 第一个 xfade: td 必须 ≤ 0.3 (第二个输入 clip_durations[1]=0.3)
+        assert durations_found[0] <= 0.3 + 0.001, (
+            f"第一个 xfade td={durations_found[0]} 超过 clip_durations[1]=0.3"
+        )
+        # 第二个 xfade: td 可以 = 0.5 (clip_durations[2]=5.0)
+        assert durations_found[1] <= 0.5 + 0.001
+        assert dur > 0
+
+    def test_offset_plus_td_never_exceeds_input(self):
+        """压力测试：多种时长组合，offset + td 永远不超过 first_input_duration。"""
+        test_cases = [
+            ([0.1, 5.0], 0.5),
+            ([0.5, 0.5], 0.5),
+            ([1.0, 1.0, 1.0], 0.5),
+            ([0.2, 0.3, 0.4], 0.5),
+            ([10.0, 0.1], 0.5),
+            ([3.0, 3.0, 3.0, 3.0], 0.5),
+            ([5.0, 0.3, 5.0], 0.5),  # P1: 中间片段短于 td
+            ([5.0, 0.1, 0.1, 5.0], 0.5),  # P1: 多个中间片段都短于 td
+        ]
+        for durations, td in test_cases:
+            labels = [f"v{i}" for i in range(len(durations))]
+            transitions = ["cut"] + ["fade"] * (len(durations) - 1)
+            f, dur = build_xfade_filter_chain(
+                clip_durations=durations,
+                clip_video_labels=labels,
+                transitions=transitions,
+                transition_duration=td,
+            )
+            assert dur >= 0, f" durations={durations} td={td} → dur={dur}"
+            # 解析 filter 验证 offset + td 的合理性
+            import re
+
+            # 按 xfade 步骤索引追踪第二个输入
+            xfade_idx = 0
+            for part in f.split(";"):
+                if "xfade=" not in part:
+                    continue
+                # 格式: [prev][next]xfade=transition=X:duration=D:offset=O[out]
+                m = re.search(
+                    r"xfade=transition=(\w+):duration=([\d.]+):offset=([\d.]+)",
+                    part,
+                )
+                assert m, f"无法解析 xfade 参数: {part}"
+                offset_val = float(m.group(3))
+                dur_val = float(m.group(2))
+                assert offset_val >= 0
+                assert dur_val >= 0.001  # 至少 1ms
+                # P1 修复验证: td 不能超过第二个输入片段时长
+                second_input_idx = xfade_idx + 1
+                assert dur_val <= durations[second_input_idx] + 0.001, (
+                    f"td={dur_val} > clip_durations[{second_input_idx}]={durations[second_input_idx]}"
+                )
+                xfade_idx += 1
+
+
+# ── P0-3: effective_duration trim 逻辑验证 ────────────────────────────────────
+
+
+class TestEffectiveDurationTrim:
+    """验证 _build_filter_complex 中 effective_duration trim 逻辑。"""
+
+    def _make_service(self, clips, asset_paths=None):
+        from pathlib import Path
+
+        from video_processing.unified_render_service import UnifiedRenderService
+
+        plan = MagicMock()
+        plan.id = "test_plan"
+        work_dir = Path("/tmp/test_render")
+        if asset_paths is None:
+            asset_paths = {}
+            for c in clips:
+                asset_paths[c.asset_id] = Path(f"/tmp/{c.asset_id}")
+        return UnifiedRenderService(
+            plan=plan,
+            clips=clips,
+            asset_path_map=asset_paths,
+            work_dir=work_dir,
+        )
+
+    def _make_clip(self, clip_id, duration=0.0, actual_duration=5.0, clip_type="main", order=0):
+        from pathlib import Path
+
+        from video_processing.unified_render_service import ResolvedClip
+
+        return ResolvedClip(
+            clip_id=clip_id,
+            asset_id=f"asset_{clip_id}.mp4",
+            local_path=Path(f"/tmp/asset_{clip_id}.mp4"),
+            clip_type=clip_type,
+            order=order,
+            duration=duration,
+            actual_duration=actual_duration,
+            transition_effect="fade",
+            config={},
+        )
+
+    def test_trim_applied_when_duration_less_than_actual(self):
+        """clip.duration < actual_duration → trim=duration=clip.duration。"""
+        clip = self._make_clip("c1", duration=3.0, actual_duration=10.0)
+        svc = self._make_service([clip])
+        layers = svc._group_clips_into_layers([clip])
+        fc, _ = svc._build_filter_complex(layers)
+        assert "trim=duration=3.0" in fc
+
+    def test_trim_uses_actual_when_no_duration_set(self):
+        """clip.duration=0 → 使用 actual_duration 做 trim。"""
+        clip = self._make_clip("c1", duration=0.0, actual_duration=7.5)
+        svc = self._make_service([clip])
+        layers = svc._group_clips_into_layers([clip])
+        fc, _ = svc._build_filter_complex(layers)
+        assert "trim=duration=7.5" in fc
+
+    def test_trim_uses_min_of_duration_and_actual(self):
+        """clip.duration > actual_duration → trim 到 actual_duration。"""
+        clip = self._make_clip("c1", duration=10.0, actual_duration=2.0)
+        svc = self._make_service([clip])
+        layers = svc._group_clips_into_layers([clip])
+        fc, _ = svc._build_filter_complex(layers)
+        assert "trim=duration=2.0" in fc
+
+    def test_no_trim_when_both_zero(self):
+        """duration=0 且 actual_duration=0 → 不做 trim。"""
+        clip = self._make_clip("c1", duration=0.0, actual_duration=0.0)
+        svc = self._make_service([clip])
+        layers = svc._group_clips_into_layers([clip])
+        fc, _ = svc._build_filter_complex(layers)
+        assert "trim=" not in fc
+
+    def test_xfade_uses_effective_durations(self):
+        """多片段 xfade 使用 trim 后的有效时长。"""
+        clip1 = self._make_clip("c1", duration=3.0, actual_duration=10.0, order=0)
+        clip2 = self._make_clip("c2", duration=4.0, actual_duration=10.0, order=1)
+        svc = self._make_service([clip1, clip2])
+        layers = svc._group_clips_into_layers([clip1, clip2])
+        fc, _ = svc._build_filter_complex(layers)
+        assert "xfade=" in fc
+        # 两个 clip 的 trim 应该分别用 3.0 和 4.0
+        assert "trim=duration=3.0" in fc
+        assert "trim=duration=4.0" in fc
+
+
+# ── P0-2: get_download_url 预签名 URL 逻辑验证 ────────────────────────────────
+
+
+class TestGetDownloadUrl:
+    """验证 OSSStorageService.get_download_url 逻辑。"""
+
+    def test_returns_signed_url_when_bucket_configured(self):
+        """bucket 已配置 → 返回签名 URL。"""
+        from app.core.storage import OSSStorageService
+
+        with patch.object(OSSStorageService, "__init__", lambda self: None):
+            svc = OSSStorageService()
+            svc.bucket = MagicMock()
+            svc.bucket.sign_url.return_value = "https://signed-url.oss.com/file.mp4?signature=xxx"
+            svc.public_url = "https://bucket.oss.com"
+
+            result = svc.get_download_url("uploads/video.mp4")
+
+            svc.bucket.sign_url.assert_called_once_with("GET", "uploads/video.mp4", 3600)
+            assert "signed-url" in result
+
+    def test_returns_raw_url_when_bucket_none(self):
+        """bucket 未配置 → 返回原始公网 URL，并记录 warning。"""
+        from app.core.storage import OSSStorageService
+
+        with patch.object(OSSStorageService, "__init__", lambda self: None):
+            svc = OSSStorageService()
+            svc.bucket = None
+            svc.public_url = "https://bucket.oss.com"
+            svc.local_url_prefix = "/generated-files"
+
+            with patch.object(svc, "_normalize_storage_key", return_value="uploads/video.mp4"):
+                result = svc.get_download_url("uploads/video.mp4")
+
+            assert result == "https://bucket.oss.com/uploads/video.mp4"
+
+    def test_local_generated_url_returned_as_is(self):
+        """本地生成文件 URL → 原样返回，不走 OSS。"""
+        from app.core.storage import OSSStorageService
+
+        with patch.object(OSSStorageService, "__init__", lambda self: None):
+            svc = OSSStorageService()
+            svc.bucket = None
+            svc.local_url_prefix = "/generated-files"
+
+            result = svc.get_download_url("/generated-files/abc123.mp4")
+            assert result == "/generated-files/abc123.mp4"
+
+    def test_normalize_strips_full_url(self):
+        """完整 URL → 提取路径部分作为 storage_key。"""
+        from app.core.storage import OSSStorageService
+
+        with patch.object(OSSStorageService, "__init__", lambda self: None):
+            svc = OSSStorageService()
+            key = svc._normalize_storage_key("https://bucket.oss-cn-hangzhou.aliyuncs.com/uploads/video.mp4")
+            assert key == "uploads/video.mp4"
+
+    def test_normalize_preserves_plain_key(self):
+        """纯路径 → 保持不变。"""
+        from app.core.storage import OSSStorageService
+
+        with patch.object(OSSStorageService, "__init__", lambda self: None):
+            svc = OSSStorageService()
+            key = svc._normalize_storage_key("uploads/video.mp4")
+            assert key == "uploads/video.mp4"
+
+    def test_sign_url_failure_falls_back(self):
+        """sign_url 异常 → 回退到原始 URL，不崩溃。"""
+        from app.core.storage import OSSStorageService
+
+        with patch.object(OSSStorageService, "__init__", lambda self: None):
+            svc = OSSStorageService()
+            svc.bucket = MagicMock()
+            svc.bucket.sign_url.side_effect = Exception("OSS error")
+            svc.public_url = "https://bucket.oss.com"
+
+            with patch.object(svc, "_normalize_storage_key", return_value="uploads/video.mp4"):
+                result = svc.get_download_url("uploads/video.mp4")
+
+            assert result == "https://bucket.oss.com/uploads/video.mp4"
+
+    def test_diagnostic_logging_on_bucket_none(self, caplog):
+        """bucket 未配置时记录 warning 日志。"""
+        from app.core.storage import OSSStorageService
+
+        with patch.object(OSSStorageService, "__init__", lambda self: None):
+            svc = OSSStorageService()
+            svc.bucket = None
+            svc.public_url = "https://bucket.oss.com"
+            svc.local_url_prefix = "/generated-files"
+
+            with patch.object(svc, "_normalize_storage_key", return_value="uploads/video.mp4"):
+                with caplog.at_level(logging.WARNING):
+                    svc.get_download_url("https://bucket.oss.com/uploads/video.mp4")
+
+            assert any("OSS bucket not configured" in r.message for r in caplog.records)
+
+    def test_diagnostic_logging_on_sign_success(self, caplog):
+        """签名成功时记录 info 日志。"""
+        from app.core.storage import OSSStorageService
+
+        with patch.object(OSSStorageService, "__init__", lambda self: None):
+            svc = OSSStorageService()
+            svc.bucket = MagicMock()
+            svc.bucket.sign_url.return_value = "https://signed.oss.com/file.mp4?sig=xxx"
+            svc.public_url = "https://bucket.oss.com"
+
+            with caplog.at_level(logging.INFO):
+                svc.get_download_url("uploads/video.mp4")
+
+            assert any("signed URL generated" in r.message for r in caplog.records)