apps/worker/video_processing/render_audio.py

@@ -268,6 +268,10 @@ def concat_main_audio(
                 "aac",
                 "-b:a",
                 "128k",
+                "-ar",
+                "48000",
+                "-ac",
+                "2",
             ]
             if trim_start > 0:
                 command.extend(["-ss", f"{trim_start:.3f}"])
@@ -299,6 +303,9 @@ def concat_main_audio(
                 if reverse_filter:
                     audio_filters.append(reverse_filter)
 
+            # aformat 归一化：统一输出格式为 48000Hz + stereo + fltp
+            audio_filters.append("aformat=sample_rates=48000:channel_layouts=stereo:sample_fmts=fltp")
+
             filter_parts: list[str] = [f"[0:a]{','.join(audio_filters)}[outa]"]
             if video_duration > 0 and final_duration < adjusted_duration:
                 filter_parts.append(f"[outa]atrim=0:{final_duration:.3f}[final_audio]")
@@ -362,6 +369,10 @@ def concat_main_audio(
             if reverse_filter:
                 audio_filters.append(reverse_filter)
 
+        # aformat 归一化：统一采样率48000Hz + 双声道stereo + fltp采样格式
+        # concat filter 要求所有输入音频参数完全一致，否则 exit=234 失败
+        audio_filters.append("aformat=sample_rates=48000:channel_layouts=stereo:sample_fmts=fltp")
+
         filter_parts.append(f"[{i}:a]{','.join(audio_filters)}[a{i}]")
 
     audio_labels = "".join(f"[a{i}]" for i in range(len(clips)))
@@ -415,19 +426,24 @@ def mix_with_independent_audio(
 
     input_idx = 0
 
+    # aformat 归一化参数（所有音频在 concat/amix 前必须统一）
+    AFORMAT = "aformat=sample_rates=48000:channel_layouts=stereo:sample_fmts=fltp"
+
     # 1. 主图层音频 concat
     if main_clips:
         for clip in main_clips:
             input_args.extend(["-i", str(clip.local_path)])
             effective_duration = clip_effective_duration(clip)
             trim_start = getattr(clip, "start_time", 0) or 0
+            clip_filters = []
             if effective_duration > 0:
-                filter_parts.append(
-                    f"[{input_idx}:a]atrim=start={trim_start:.3f}:duration={effective_duration:.3f},"
-                    f"asetpts=PTS-STARTPTS[ma{input_idx}]"
-                )
+                clip_filters.append(f"atrim=start={trim_start:.3f}:duration={effective_duration:.3f}")
+                clip_filters.append("asetpts=PTS-STARTPTS")
             else:
-                filter_parts.append(f"[{input_idx}:a]asetpts=PTS-STARTPTS[ma{input_idx}]")
+                clip_filters.append("asetpts=PTS-STARTPTS")
+            # aformat 归一化：concat/amix 前统一音频参数，否则不同采样率/声道会失败
+            clip_filters.append(AFORMAT)
+            filter_parts.append(f"[{input_idx}:a]{','.join(clip_filters)}[ma{input_idx}]")
             input_idx += 1
 
         if len(main_clips) == 1:
@@ -450,6 +466,8 @@ def mix_with_independent_audio(
         filters.append("asetpts=PTS-STARTPTS")
         if volume != 1.0:
             filters.append(f"volume={volume}")
+        # aformat 归一化：amix 前统一音频参数
+        filters.append(AFORMAT)
         filter_parts.append(f"[{input_idx}:a]{','.join(filters)}[{label}]")
         mix_labels.append(label)
         input_idx += 1

apps/worker/video_processing/unified_render_service.py

@@ -996,15 +996,19 @@ class UnifiedRenderService:
             if reverse_filter:
                 filters.append(reverse_filter)
 
-        # scale + crop（铺满裁剪）
+        # scale + pad（等比缩放+留黑边）
         if role in ("overlay", "corner_voice"):
             pip_w = int(self.output_width * _PIP_SCALE)
             pip_h = int(self.output_height * _PIP_SCALE)
             filters.append(f"scale={pip_w}:{pip_h}")
-        else:
-            # main / broll / background: 铺满裁剪
+        elif role == "background":
+            # background: 铺满裁剪（作为底图，覆盖全屏）
             filters.append(f"scale={self.output_width}:{self.output_height}" ":force_original_aspect_ratio=increase")
             filters.append(f"crop={self.output_width}:{self.output_height}")
+        else:
+            # main / broll: 等比缩放 + 居中留黑边（保持原始比例，不裁剪内容）
+            filters.append(f"scale={self.output_width}:{self.output_height}" ":force_original_aspect_ratio=decrease")
+            filters.append(f"pad={self.output_width}:{self.output_height}:(ow-iw)/2:(oh-ih)/2:black")
 
         # 调色滤镜
         color_grade = ColorGradeConfig.from_dict(clip.config.get("color_grade"))
@@ -1333,12 +1337,12 @@ class UnifiedRenderService:
                 )
                 filters.append(f"crop={self.output_width}:{self.output_height}")
             else:
-                # main / broll: 铺满裁剪（scale to cover + center crop）
-                # 对齐链路A编辑器合成行为，与主流短视频平台一致
+                # main / broll: 等比缩放 + 居中留黑边（保持原始比例，不裁剪内容）
+                # concat 要求所有输入分辨率完全一致，pad 模式确保不同宽高比的素材都能正常拼接
                 filters.append(
-                    f"scale={self.output_width}:{self.output_height}" ":force_original_aspect_ratio=increase"
+                    f"scale={self.output_width}:{self.output_height}" ":force_original_aspect_ratio=decrease"
                 )
-                filters.append(f"crop={self.output_width}:{self.output_height}")
+                filters.append(f"pad={self.output_width}:{self.output_height}:(ow-iw)/2:(oh-ih)/2:black")
 
             # 调色滤镜（每个 clip 独立的 color grade 配置）
             color_grade = ColorGradeConfig.from_dict(clip.config.get("color_grade"))

tests/unit/test_unified_render_service.py

@@ -85,6 +85,9 @@ def _make_service(
     clips: list[FakeClip] | None = None,
     asset_paths: dict[str, Path] | None = None,
     work_dir: Path | None = None,
+    output_fps: int = 25,
+    output_width: int = 1280,
+    output_height: int = 720,
 ) -> UnifiedRenderService:
     """创建测试用的 UnifiedRenderService 实例。
 
@@ -104,6 +107,9 @@ def _make_service(
         clips=clips,
         asset_path_map=asset_paths,
         work_dir=work_dir,
+        output_width=output_width,
+        output_height=output_height,
+        output_fps=output_fps,
     )
 
 
@@ -486,10 +492,10 @@ class TestBuildFilterComplex:
             f"单视频: setpts(position={last_setpts}) 应该在 fps(position={first_fps}) 之前。" f"滤镜链: {chain_str}"
         )
 
-    def test_main_clip_uses_fill_crop_strategy(self):
-        """main/broll clip 使用铺满裁剪策略（scale increase + crop），不是等比+黑边。
+    def test_main_clip_uses_pad_strategy(self):
+        """main clip 使用 scale+pad 保持比例留黑边（不裁剪内容）。
 
-        对齐链路A编辑器合成行为，与主流短视频平台一致。
+        渲染引擎不能挑素材，任何素材进来都统一规格后出片；pad留黑边保证内容完整。
         """
         clips = [_make_clip("c1", "main", order=0, duration=5.0)]
         asset_paths = {"asset_c1.mp4": Path("/tmp/asset_c1.mp4")}
@@ -500,15 +506,16 @@ class TestBuildFilterComplex:
             layers = svc._group_clips_into_layers(resolved)
             fc, _ = svc._build_filter_complex(layers)
 
-        # 验证：scale 使用 force_original_aspect_ratio=increase（铺满）
-        assert "force_original_aspect_ratio=increase" in fc
-        # 验证：有 crop（居中裁剪）
-        assert "crop=1280:720" in fc
-        # 验证：没有 pad（不是黑边模式）
-        assert "pad=" not in fc
+        # 验证：scale 使用 force_original_aspect_ratio=decrease（等比缩小）
+        assert "force_original_aspect_ratio=decrease" in fc
+        # 验证：有 pad（留黑边）
+        assert "pad=" in fc
+        assert "black" in fc
+        # 验证：没有 crop（不是裁剪模式）
+        assert "crop=" not in fc
 
-    def test_broll_clip_uses_fill_crop_strategy(self):
-        """broll clip 同样使用铺满裁剪策略。"""
+    def test_broll_clip_uses_pad_strategy(self):
+        """broll clip 同样使用 scale+pad 留黑边策略。"""
         clips = [_make_clip("c1", "b_roll", order=0, duration=5.0)]
         asset_paths = {"asset_c1.mp4": Path("/tmp/asset_c1.mp4")}
         svc = _make_service(clips, asset_paths)
@@ -518,9 +525,9 @@ class TestBuildFilterComplex:
             layers = svc._group_clips_into_layers(resolved)
             fc, _ = svc._build_filter_complex(layers)
 
-        assert "force_original_aspect_ratio=increase" in fc
-        assert "crop=1280:720" in fc
-        assert "pad=" not in fc
+        assert "force_original_aspect_ratio=decrease" in fc
+        assert "pad=" in fc
+        assert "crop=" not in fc
 
     def test_background_uses_fill_crop_strategy(self):
         """background 层也使用铺满裁剪（已有的行为，保持一致）。"""
@@ -1765,3 +1772,493 @@ class TestExtractAudioUsesRunFfmpeg:
             mock_run.side_effect = subprocess.CalledProcessError(1, "ffmpeg", stderr="error")
             with pytest.raises(RuntimeError, match="音频提取失败"):
                 svc._extract_audio(video_path, output_path)
+
+
+# ══════════════════════════════════════════════════════════════════════════════
+# concat 前 4 项归一化测试（修复 exit=234）
+# ══════════════════════════════════════════════════════════════════════════════
+
+
+class TestConcatNormalizeVideoResolution:
+    """视频分辨率归一化：scale + pad 保持比例留黑边，确保 concat 前分辨率一致。
+
+    对应 4 项 normalize 之：视频分辨率（scale + pad 到 canvas_width × canvas_height）
+    修复前：main/broll 用 scale+crop（铺满裁剪），横屏素材内容被裁掉
+    修复后：main/broll 用 scale+pad（留黑边），保持原始比例不裁剪
+    """
+
+    def test_main_layer_uses_pad_not_crop_in_filter_complex(self):
+        """_build_filter_complex 中 main 层用 scale+pad（letterbox），不用 crop。"""
+        clips = [_make_clip("c1", "main", order=0, duration=3.0)]
+        asset_paths = {"asset_c1.mp4": Path("/tmp/asset_c1.mp4")}
+        svc = _make_service(clips, asset_paths)
+
+        with _patch_path_exists(), patch("video_processing.unified_render_service.probe_duration", return_value=5.0):
+            resolved = svc._resolve_clips()
+            layers = svc._group_clips_into_layers(resolved)
+            fc, _ = svc._build_filter_complex(layers)
+
+        # 必须用 pad（留黑边），不能用 crop（裁剪）
+        assert "force_original_aspect_ratio=decrease" in fc, "main 层应使用 decrease 模式（等比缩放到画布内）"
+        assert "pad=" in fc, "main 层应有 pad 滤镜（留黑边到目标分辨率）"
+        assert "(ow-iw)/2:(oh-ih)/2:black" in fc, "pad 应该居中+黑边"
+        # 注意：background 层也用 crop，但 main/broll 应该用 pad
+        # 检查第一个 [0:v] 处理链（第一个 clip 是 main）
+        first_v_chain = fc.split("[v0]")[0]
+        assert "crop=" not in first_v_chain, "main 层不应该有 crop 滤镜"
+
+    def test_broll_layer_uses_pad_not_crop_in_filter_complex(self):
+        """_build_filter_complex 中 b_roll 层用 scale+pad，不用 crop。"""
+        clips = [
+            _make_clip("c1", "b_roll", order=0, duration=3.0),
+        ]
+        asset_paths = {"asset_c1.mp4": Path("/tmp/asset_c1.mp4")}
+        svc = _make_service(clips, asset_paths)
+
+        with _patch_path_exists(), patch("video_processing.unified_render_service.probe_duration", return_value=5.0):
+            resolved = svc._resolve_clips()
+            layers = svc._group_clips_into_layers(resolved)
+            fc, _ = svc._build_filter_complex(layers)
+
+        assert "force_original_aspect_ratio=decrease" in fc
+        assert "pad=" in fc
+        first_v_chain = fc.split("[v0]")[0]
+        assert "crop=" not in first_v_chain, "broll 层不应该有 crop 滤镜"
+
+    def test_background_layer_still_uses_crop_in_filter_complex(self):
+        """_build_filter_complex 中 background 层保持 scale+crop（作为底图铺满）。"""
+        clips = [
+            _make_clip("bg1", "background", order=0, duration=3.0),
+            _make_clip("c1", "main", order=0, duration=3.0),
+        ]
+        asset_paths = {
+            "asset_bg1.mp4": Path("/tmp/asset_bg1.mp4"),
+            "asset_c1.mp4": Path("/tmp/asset_c1.mp4"),
+        }
+        svc = _make_service(clips, asset_paths)
+
+        with _patch_path_exists(), patch("video_processing.unified_render_service.probe_duration", return_value=5.0):
+            resolved = svc._resolve_clips()
+            layers = svc._group_clips_into_layers(resolved)
+            fc, _ = svc._build_filter_complex(layers)
+
+        # background 应该用 increase + crop（cover 模式）
+        # 统计 crop 出现次数：background 1次 + 其他位置（如果有）
+        # 关键是 background 走 increase 模式
+        assert "force_original_aspect_ratio=increase" in fc, "background 层应使用 increase 模式（铺满裁剪）"
+
+    def test_multi_clip_concat_all_same_resolution(self):
+        """多 clip concat 时，所有 clip 预处理后分辨率一致（pad 到同一尺寸）。"""
+        clips = [
+            _make_clip("c1", "main", order=0, duration=3.0),
+            _make_clip("c2", "main", order=1, duration=2.0),
+        ]
+        asset_paths = {
+            "asset_c1.mp4": Path("/tmp/asset_c1.mp4"),
+            "asset_c2.mp4": Path("/tmp/asset_c2.mp4"),
+        }
+        svc = _make_service(clips, asset_paths)
+
+        with _patch_path_exists(), patch("video_processing.unified_render_service.probe_duration", return_value=5.0):
+            resolved = svc._resolve_clips()
+            layers = svc._group_clips_into_layers(resolved)
+            fc, _ = svc._build_filter_complex(layers)
+
+        # 两个 clip 都应该有 pad
+        pad_count = fc.count("pad=")
+        assert pad_count >= 2, f"至少应有 2 个 pad（每个 clip 一个），实际 {pad_count}"
+        # 都用 decrease 模式
+        decrease_count = fc.count("force_original_aspect_ratio=decrease")
+        assert decrease_count >= 2, f"至少应有 2 个 decrease，实际 {decrease_count}"
+        # 有 concat
+        assert "concat=n=2:v=1:a=0" in fc
+
+    def test_pass_through_main_uses_pad_not_crop(self):
+        """_render_pass_through 中 main 层用 scale+pad，不用 crop。"""
+        clips = [_make_clip("c1", "main", order=0, duration=3.0)]
+        asset_paths = {"asset_c1.mp4": Path("/tmp/asset_c1.mp4")}
+        svc = _make_service(clips, asset_paths)
+
+        with (
+            _patch_path_exists(),
+            patch("video_processing.unified_render_service.probe_duration", return_value=5.0),
+            patch("video_processing.unified_render_service.run_ffmpeg") as mock_run,
+        ):
+            resolved = svc._resolve_clips()
+            layers = svc._group_clips_into_layers(resolved)
+            svc._render_pass_through(layers, Path("/tmp/out.mp4"), video_duration=3.0)
+
+        assert mock_run.called
+        cmd = mock_run.call_args[0][0]
+        cmd_str = " ".join(cmd)
+
+        # 直通模式 main 层应该用 pad
+        assert "force_original_aspect_ratio=decrease" in cmd_str, "直通模式 main 层应使用 decrease 模式"
+        assert "pad=" in cmd_str, "直通模式应有 pad 滤镜"
+        # 检查 vf 中没有 crop
+        vf_idx = cmd.index("-vf")
+        vf_value = cmd[vf_idx + 1]
+        assert "crop=" not in vf_value, "直通模式 main 层 vf 中不应该有 crop"
+
+    def test_pass_through_background_uses_crop(self):
+        """_render_pass_through 中 background 层保持 scale+crop。"""
+        clips = [_make_clip("bg1", "background", order=0, duration=3.0)]
+        asset_paths = {"asset_bg1.mp4": Path("/tmp/asset_bg1.mp4")}
+        svc = _make_service(clips, asset_paths)
+
+        with (
+            _patch_path_exists(),
+            patch("video_processing.unified_render_service.probe_duration", return_value=5.0),
+            patch("video_processing.unified_render_service.run_ffmpeg") as mock_run,
+        ):
+            resolved = svc._resolve_clips()
+            layers = svc._group_clips_into_layers(resolved)
+            svc._render_pass_through(layers, Path("/tmp/out.mp4"), video_duration=3.0)
+
+        assert mock_run.called
+        cmd = mock_run.call_args[0][0]
+        cmd_str = " ".join(cmd)
+
+        # background 层应该用 increase + crop
+        assert "force_original_aspect_ratio=increase" in cmd_str
+        vf_idx = cmd.index("-vf")
+        vf_value = cmd[vf_idx + 1]
+        assert "crop=" in vf_value, "background 层应有 crop 滤镜"
+
+
+class TestConcatNormalizeVideoFps:
+    """视频帧率归一化：fps 滤镜统一到目标 fps。
+
+    对应 4 项 normalize 之：视频帧率（fps 滤镜统一到目标 fps）
+    """
+
+    def test_filter_complex_has_fps_filter(self):
+        """_build_filter_complex 中每个 clip 都有 fps 滤镜。"""
+        clips = [
+            _make_clip("c1", "main", order=0, duration=3.0),
+            _make_clip("c2", "main", order=1, duration=2.0),
+        ]
+        asset_paths = {
+            "asset_c1.mp4": Path("/tmp/asset_c1.mp4"),
+            "asset_c2.mp4": Path("/tmp/asset_c2.mp4"),
+        }
+        svc = _make_service(clips, asset_paths, output_fps=30)
+
+        with _patch_path_exists(), patch("video_processing.unified_render_service.probe_duration", return_value=5.0):
+            resolved = svc._resolve_clips()
+            layers = svc._group_clips_into_layers(resolved)
+            fc, _ = svc._build_filter_complex(layers)
+
+        fps_count = fc.count("fps=30")
+        assert fps_count >= 2, f"至少应有 2 个 fps=30（每个 clip 一个），实际 {fps_count}"
+
+    def test_fps_after_pad_before_final_setpts(self):
+        """fps 滤镜在 pad 之后、setpts 之前（确保分辨率和帧率都统一后再归一化 PTS）。"""
+        import re
+
+        clips = [_make_clip("c1", "main", order=0, duration=3.0)]
+        asset_paths = {"asset_c1.mp4": Path("/tmp/asset_c1.mp4")}
+        svc = _make_service(clips, asset_paths, output_fps=30)
+
+        with _patch_path_exists(), patch("video_processing.unified_render_service.probe_duration", return_value=5.0):
+            resolved = svc._resolve_clips()
+            layers = svc._group_clips_into_layers(resolved)
+            fc, _ = svc._build_filter_complex(layers)
+
+        # 提取第一个 clip 的视频处理链
+        chain_str = fc.split("[v0]")[0]
+
+        pad_pos = chain_str.find("pad=")
+        fps_pos = chain_str.find("fps=")
+        assert pad_pos >= 0, "应找到 pad 滤镜"
+        assert fps_pos >= 0, "应找到 fps 滤镜"
+        assert fps_pos > pad_pos, "fps 应在 pad 之后"
+
+
+class TestConcatNormalizeAudioFormat:
+    """音频格式归一化：aformat 统一为 stereo + 48000Hz + fltp。
+
+    对应 4 项 normalize 之：音频声道（aformat 统一为 stereo + 48000Hz + fltp）
+    修复前：concat 前音频无归一化，不同采样率/声道导致 exit=234
+    修复后：concat/amix 前所有音频统一为 48000Hz + stereo + fltp
+    """
+
+    def test_multi_clip_concat_has_aformat(self):
+        """多 clip 音频 concat 前，每个 clip 都有 aformat 归一化。"""
+        clips = [
+            _make_clip("c1", "main", order=0, duration=3.0),
+            _make_clip("c2", "main", order=1, duration=2.0),
+        ]
+        asset_paths = {
+            "asset_c1.mp4": Path("/tmp/asset_c1.mp4"),
+            "asset_c2.mp4": Path("/tmp/asset_c2.mp4"),
+        }
+        svc = _make_service(clips, asset_paths)
+
+        with (
+            _patch_path_exists(),
+            patch("video_processing.unified_render_service.probe_duration", return_value=5.0),
+            patch("video_processing.render_audio.run_ffmpeg") as mock_run,
+        ):
+            layers = svc._group_clips_into_layers(svc._resolve_clips())
+            ctx = _make_ctx()
+            mix_audio(ctx, layers, 4.5)
+
+        assert mock_run.called
+        cmd = mock_run.call_args[0][0]
+        cmd_str = " ".join(cmd)
+
+        # 必须有 aformat 归一化
+        assert "aformat=" in cmd_str, "concat 前应有 aformat 滤镜"
+        assert "sample_rates=48000" in cmd_str, "采样率应统一为 48000Hz"
+        assert "channel_layouts=stereo" in cmd_str, "声道应统一为 stereo"
+        assert "sample_fmts=fltp" in cmd_str, "采样格式应统一为 fltp"
+
+        # 两个 clip 都应该有 aformat
+        aformat_count = cmd_str.count("aformat=")
+        assert aformat_count >= 2, f"每个 clip 都应有 aformat，实际 {aformat_count} 个"
+
+        # 有 concat
+        assert "concat=n=2:v=0:a=1" in cmd_str
+
+    def test_aformat_before_concat(self):
+        """aformat 应在 concat 之前（每个 clip 处理链中 aformat 在 concat 之前）。"""
+        clips = [
+            _make_clip("c1", "main", order=0, duration=3.0),
+            _make_clip("c2", "main", order=1, duration=2.0),
+        ]
+        asset_paths = {
+            "asset_c1.mp4": Path("/tmp/asset_c1.mp4"),
+            "asset_c2.mp4": Path("/tmp/asset_c2.mp4"),
+        }
+        svc = _make_service(clips, asset_paths)
+
+        with (
+            _patch_path_exists(),
+            patch("video_processing.unified_render_service.probe_duration", return_value=5.0),
+            patch("video_processing.render_audio.run_ffmpeg") as mock_run,
+        ):
+            layers = svc._group_clips_into_layers(svc._resolve_clips())
+            ctx = _make_ctx()
+            mix_audio(ctx, layers, 4.5)
+
+        cmd = mock_run.call_args[0][0]
+        fc_idx = cmd.index("-filter_complex")
+        fc_str = cmd[fc_idx + 1]
+
+        # 找到 concat 的位置
+        concat_pos = fc_str.find("concat=")
+        assert concat_pos > 0, "应找到 concat filter"
+
+        # 在 concat 之前的部分应该有 aformat
+        before_concat = fc_str[:concat_pos]
+        aformat_before_count = before_concat.count("aformat=")
+        assert aformat_before_count >= 2, f"concat 之前每个 clip 都应有 aformat，实际 {aformat_before_count} 个"
+
+    def test_single_clip_audio_has_normalized_output(self):
+        """单 clip 音频输出也应统一格式（一致性保障）。"""
+        clips = [_make_clip("c1", "main", order=0, duration=5.0)]
+        asset_paths = {"asset_c1.mp4": Path("/tmp/asset_c1.mp4")}
+        svc = _make_service(clips, asset_paths)
+
+        with (
+            _patch_path_exists(),
+            patch("video_processing.unified_render_service.probe_duration", return_value=5.0),
+            patch("video_processing.render_audio.run_ffmpeg") as mock_run,
+        ):
+            layers = svc._group_clips_into_layers(svc._resolve_clips())
+            ctx = _make_ctx()
+            mix_audio(ctx, layers, 5.0)
+
+        assert mock_run.called
+        cmd = mock_run.call_args[0][0]
+
+        # 单 clip 简单路径应该有 -ar 48000 和 -ac 2
+        assert "-ar" in cmd, "单 clip 音频应指定采样率"
+        ar_idx = cmd.index("-ar")
+        assert cmd[ar_idx + 1] == "48000", "采样率应为 48000"
+        assert "-ac" in cmd, "单 clip 音频应指定声道数"
+        ac_idx = cmd.index("-ac")
+        assert cmd[ac_idx + 1] == "2", "声道数应为 2（stereo）"
+
+    def test_independent_audio_track_has_aformat(self):
+        """独立音频轨在 amix 前也有 aformat 归一化。"""
+        clips = [
+            _make_clip("c1", "main", order=0, duration=5.0),
+            _make_clip(
+                "audio1",
+                "main",
+                order=0,
+                duration=5.0,
+                config={"role": "audio", "volume": 0.5},
+            ),
+        ]
+        asset_paths = {
+            "asset_c1.mp4": Path("/tmp/asset_c1.mp4"),
+            "asset_audio1.mp4": Path("/tmp/asset_audio1.mp4"),
+        }
+        svc = _make_service(clips, asset_paths)
+
+        with (
+            _patch_path_exists(),
+            patch("video_processing.unified_render_service.probe_duration", return_value=5.0),
+            patch("video_processing.render_audio.run_ffmpeg") as mock_run,
+        ):
+            layers = svc._group_clips_into_layers(svc._resolve_clips())
+            ctx = _make_ctx()
+            mix_audio(ctx, layers, 5.0)
+
+        assert mock_run.called
+        cmd = mock_run.call_args[0][0]
+        cmd_str = " ".join(cmd)
+
+        # 应该有 aformat
+        assert "aformat=" in cmd_str
+        assert "sample_rates=48000" in cmd_str
+        assert "channel_layouts=stereo" in cmd_str
+        # amix 应该存在
+        assert "amix" in cmd_str
+
+
+class TestConcatNormalizeAudioCodec:
+    """音频编码归一化：统一为 aac。
+
+    对应 4 项 normalize 之：音频编码（统一为 aac）
+    """
+
+    def test_multi_clip_output_is_aac(self):
+        """多 clip concat 后输出编码为 aac。"""
+        clips = [
+            _make_clip("c1", "main", order=0, duration=3.0),
+            _make_clip("c2", "main", order=1, duration=2.0),
+        ]
+        asset_paths = {
+            "asset_c1.mp4": Path("/tmp/asset_c1.mp4"),
+            "asset_c2.mp4": Path("/tmp/asset_c2.mp4"),
+        }
+        svc = _make_service(clips, asset_paths)
+
+        with (
+            _patch_path_exists(),
+            patch("video_processing.unified_render_service.probe_duration", return_value=5.0),
+            patch("video_processing.render_audio.run_ffmpeg") as mock_run,
+        ):
+            layers = svc._group_clips_into_layers(svc._resolve_clips())
+            ctx = _make_ctx()
+            mix_audio(ctx, layers, 4.5)
+
+        assert mock_run.called
+        cmd = mock_run.call_args[0][0]
+        assert "aac" in cmd, "音频输出编码应为 aac"
+        assert "-b:a" in cmd, "应指定音频码率"
+
+    def test_single_clip_output_is_aac(self):
+        """单 clip 音频输出编码为 aac。"""
+        clips = [_make_clip("c1", "main", order=0, duration=5.0)]
+        asset_paths = {"asset_c1.mp4": Path("/tmp/asset_c1.mp4")}
+        svc = _make_service(clips, asset_paths)
+
+        with (
+            _patch_path_exists(),
+            patch("video_processing.unified_render_service.probe_duration", return_value=5.0),
+            patch("video_processing.render_audio.run_ffmpeg") as mock_run,
+        ):
+            layers = svc._group_clips_into_layers(svc._resolve_clips())
+            ctx = _make_ctx()
+            mix_audio(ctx, layers, 5.0)
+
+        assert mock_run.called
+        cmd = mock_run.call_args[0][0]
+        assert "aac" in cmd
+
+
+class TestConcatNormalizeFourItemsComplete:
+    """4 项 normalize 完整性验证：one_take 多素材 concat 场景全链路检查。
+
+    模拟 one_take 模式（多个 main clip）的完整渲染路径，
+    验证 concat 前所有 4 项 normalize 都已到位。
+    """
+
+    def _make_one_take_clips(self):
+        """构造 one_take 模式的典型素材：3 个 main clip（模拟横屏+竖屏混合）。"""
+        return [
+            _make_clip("c1", "main", order=0, duration=5.0),
+            _make_clip("c2", "main", order=1, duration=4.0),
+            _make_clip("c3", "main", order=2, duration=3.0),
+        ]
+
+    def test_video_resolution_normalized(self):
+        """[1/4] 视频分辨率：所有 clip 都有 pad 到统一分辨率。"""
+        clips = self._make_one_take_clips()
+        asset_paths = {f"asset_c{i}.mp4": Path(f"/tmp/asset_c{i}.mp4") for i in range(1, 4)}
+        svc = _make_service(clips, asset_paths)
+
+        with _patch_path_exists(), patch("video_processing.unified_render_service.probe_duration", return_value=10.0):
+            resolved = svc._resolve_clips()
+            layers = svc._group_clips_into_layers(resolved)
+            fc, _ = svc._build_filter_complex(layers)
+
+        # 3 个 clip 都应该有 pad（decrease + black padding）
+        pad_count = fc.count("pad=")
+        decrease_count = fc.count("force_original_aspect_ratio=decrease")
+        assert pad_count >= 3, f"3个 clip 都应有 pad，实际 {pad_count} 个"
+        assert decrease_count >= 3, f"3个 clip 都应用 decrease 模式，实际 {decrease_count} 个"
+
+    def test_video_fps_normalized(self):
+        """[2/4] 视频帧率：所有 clip 都有 fps 滤镜。"""
+        clips = self._make_one_take_clips()
+        asset_paths = {f"asset_c{i}.mp4": Path(f"/tmp/asset_c{i}.mp4") for i in range(1, 4)}
+        svc = _make_service(clips, asset_paths, output_fps=30)
+
+        with _patch_path_exists(), patch("video_processing.unified_render_service.probe_duration", return_value=10.0):
+            resolved = svc._resolve_clips()
+            layers = svc._group_clips_into_layers(resolved)
+            fc, _ = svc._build_filter_complex(layers)
+
+        fps_count = fc.count("fps=30")
+        assert fps_count >= 3, f"3个 clip 都应有 fps=30，实际 {fps_count} 个"
+
+    def test_audio_format_normalized(self):
+        """[3/4] 音频格式：所有 clip concat 前都有 aformat 归一化。"""
+        clips = self._make_one_take_clips()
+        asset_paths = {f"asset_c{i}.mp4": Path(f"/tmp/asset_c{i}.mp4") for i in range(1, 4)}
+        svc = _make_service(clips, asset_paths)
+
+        with (
+            _patch_path_exists(),
+            patch("video_processing.unified_render_service.probe_duration", return_value=10.0),
+            patch("video_processing.render_audio.run_ffmpeg") as mock_run,
+        ):
+            layers = svc._group_clips_into_layers(svc._resolve_clips())
+            ctx = _make_ctx()
+            mix_audio(ctx, layers, 12.0)
+
+        assert mock_run.called
+        cmd = mock_run.call_args[0][0]
+        cmd_str = " ".join(cmd)
+
+        aformat_count = cmd_str.count("aformat=")
+        assert aformat_count >= 3, f"3个 clip 都应有 aformat，实际 {aformat_count} 个"
+        assert "sample_rates=48000" in cmd_str
+        assert "channel_layouts=stereo" in cmd_str
+        assert "sample_fmts=fltp" in cmd_str
+
+    def test_audio_codec_aac(self):
+        """[4/4] 音频编码：输出为 aac。"""
+        clips = self._make_one_take_clips()
+        asset_paths = {f"asset_c{i}.mp4": Path(f"/tmp/asset_c{i}.mp4") for i in range(1, 4)}
+        svc = _make_service(clips, asset_paths)
+
+        with (
+            _patch_path_exists(),
+            patch("video_processing.unified_render_service.probe_duration", return_value=10.0),
+            patch("video_processing.render_audio.run_ffmpeg") as mock_run,
+        ):
+            layers = svc._group_clips_into_layers(svc._resolve_clips())
+            ctx = _make_ctx()
+            mix_audio(ctx, layers, 12.0)
+
+        assert mock_run.called
+        cmd = mock_run.call_args[0][0]
+        assert "aac" in cmd
+        assert "concat=n=3:v=0:a=1" in " ".join(cmd)