Merge pull request #266 from 0xDevNinja/fix/audio-mixer-ducking-filtergraph

fix: three independent zero-key correctness bugs (audio_mixer ducking, scene-variation run count, source-media schema)

lib/variation_checker.py

@@ -56,13 +56,20 @@ def check_scene_variation(scenes: list[dict[str, Any]]) -> dict[str, Any]:
             suggestions.append("Mix wide establishing shots with close-ups for visual rhythm.")
 
     # --- Check 2: Consecutive same-size shots ---
-    consecutive_same = 0
+    # Track the longest actual run of identical shot sizes. Summing every equal
+    # adjacent pair across the whole plan would count non-consecutive groups
+    # (e.g. wide,wide,cu,cu,med,med -> 3 pairs) as a single "3 consecutive" run.
+    longest_run = 1 if shot_sizes else 0
+    current_run = 1
     for i in range(1, len(shot_sizes)):
         if shot_sizes[i] == shot_sizes[i-1] and shot_sizes[i] != "unspecified":
-            consecutive_same += 1
-    if consecutive_same >= 3:
+            current_run += 1
+            longest_run = max(longest_run, current_run)
+        else:
+            current_run = 1
+    if longest_run >= 3:
         violations.append(
-            f"{consecutive_same} consecutive same-size shots. "
+            f"{longest_run} consecutive same-size shots. "
             f"Vary shot sizes between scenes for editorial rhythm."
         )
 

schemas/artifacts/source_media_review.schema.json

@@ -9,6 +9,7 @@
     "version": { "type": "string", "const": "1.0" },
     "files": {
       "type": "array",
+      "description": "Reviewed source files. Empty when no user media was supplied (or none could be reviewed) — a valid 'fully generated production' state that review_source_media reports explicitly.",
       "items": {
         "type": "object",
         "required": ["path", "media_type", "reviewed"],
@@ -64,7 +65,7 @@
         },
         "additionalProperties": false
       },
-      "minItems": 1
+      "minItems": 0
     },
     "summary": {
       "type": "string",

tests/lib/test_source_media_review_empty.py

@@ -0,0 +1,29 @@
+"""Regression test for source_media_review empty-files artifact validity.
+
+review_source_media deliberately returns an artifact with files:[] when no user
+media was supplied (or none could be reviewed) — a valid "fully generated
+production" state. The schema declared files.minItems: 1, so that intended
+artifact failed its own schema validation.
+"""
+
+import sys
+from pathlib import Path
+
+PROJECT_ROOT = Path(__file__).resolve().parent.parent.parent
+sys.path.insert(0, str(PROJECT_ROOT))
+
+from lib.source_media_review import review_source_media  # noqa: E402
+from schemas.artifacts import validate_artifact  # noqa: E402
+
+
+def test_no_source_media_produces_schema_valid_artifact(tmp_path):
+    art = review_source_media([tmp_path / "does-not-exist.mp4"], {})
+    assert art["files"] == []
+    # Must not raise — this is a legitimate no-source-media artifact.
+    validate_artifact("source_media_review", art)
+
+
+def test_no_files_at_all_is_schema_valid():
+    art = review_source_media([], {})
+    assert art["files"] == []
+    validate_artifact("source_media_review", art)

tests/lib/test_variation_checker_runs.py

@@ -0,0 +1,37 @@
+"""Regression test for check_scene_variation consecutive-run counting.
+
+The "consecutive same-size shots" check summed every equal adjacent pair across
+the whole plan instead of measuring the longest actual run, so an editorially
+varied plan of separate 2-shot groups (wide,wide,cu,cu,med,med) falsely tripped
+a "3 consecutive same-size shots" violation.
+"""
+
+import sys
+from pathlib import Path
+
+PROJECT_ROOT = Path(__file__).resolve().parent.parent.parent
+sys.path.insert(0, str(PROJECT_ROOT))
+
+from lib.variation_checker import check_scene_variation  # noqa: E402
+
+
+def _scenes(sizes):
+    return [{"shot_language": {"shot_size": s}} for s in sizes]
+
+
+def test_non_consecutive_same_size_pairs_do_not_trip_run_check():
+    # Three separate 2-shot groups — longest run is 2, not 3.
+    res = check_scene_variation(_scenes(["wide", "wide", "cu", "cu", "medium", "medium"]))
+    assert not any("consecutive same-size" in v for v in res["violations"])
+
+
+def test_true_run_of_three_is_flagged():
+    res = check_scene_variation(_scenes(["wide", "wide", "wide", "cu", "medium"]))
+    assert any("3 consecutive same-size" in v for v in res["violations"])
+
+
+def test_unspecified_shots_do_not_form_a_run():
+    res = check_scene_variation(
+        _scenes(["unspecified", "unspecified", "unspecified", "unspecified"])
+    )
+    assert not any("consecutive same-size" in v for v in res["violations"])

tests/tools/test_audio_mixer_ducking.py

@@ -0,0 +1,89 @@
+"""Regression tests for audio_mixer full_mix ducking filtergraph.
+
+The ducking branch built an `acopy[speech_dup]` filter whose output pad was
+never consumed, leaving the FFmpeg filtergraph with a dangling output. FFmpeg
+rejects that, so `full_mix` with the most common shape — a single narration
+track plus one music bed, with ducking enabled (the default) — always failed.
+"""
+
+import shutil
+import subprocess
+import sys
+from pathlib import Path
+
+import pytest
+
+PROJECT_ROOT = Path(__file__).resolve().parent.parent.parent
+sys.path.insert(0, str(PROJECT_ROOT))
+
+from tools.audio.audio_mixer import AudioMixer  # noqa: E402
+
+pytestmark = pytest.mark.skipif(
+    shutil.which("ffmpeg") is None, reason="ffmpeg required for full_mix"
+)
+
+
+def _sine(path: Path, freq: int, dur: int) -> None:
+    subprocess.run(
+        ["ffmpeg", "-y", "-f", "lavfi", "-i", f"sine=frequency={freq}:duration={dur}", str(path)],
+        capture_output=True,
+        check=True,
+        timeout=30,
+    )
+
+
+def _has_audio(path: Path) -> bool:
+    out = subprocess.run(
+        ["ffprobe", "-v", "error", "-select_streams", "a",
+         "-show_entries", "stream=codec_type", "-of", "csv=p=0", str(path)],
+        capture_output=True, text=True, timeout=30,
+    )
+    return "audio" in out.stdout
+
+
+def test_full_mix_single_narration_plus_music_with_ducking(tmp_path):
+    speech = tmp_path / "speech.wav"
+    music = tmp_path / "music.wav"
+    _sine(speech, 440, 2)
+    _sine(music, 220, 3)
+    out = tmp_path / "mixed.wav"
+
+    result = AudioMixer().execute(
+        {
+            "operation": "full_mix",
+            "tracks": [
+                {"path": str(speech), "role": "speech"},
+                {"path": str(music), "role": "music"},
+            ],
+            "ducking": {"enabled": True},
+            "output_path": str(out),
+        }
+    )
+
+    assert result.success is True, result.error
+    assert out.exists() and _has_audio(out)
+
+
+def test_full_mix_multi_narration_plus_music_with_ducking(tmp_path):
+    s1, s2 = tmp_path / "s1.wav", tmp_path / "s2.wav"
+    music = tmp_path / "music.wav"
+    _sine(s1, 440, 2)
+    _sine(s2, 330, 2)
+    _sine(music, 220, 3)
+    out = tmp_path / "mixed_multi.wav"
+
+    result = AudioMixer().execute(
+        {
+            "operation": "full_mix",
+            "tracks": [
+                {"path": str(s1), "role": "speech"},
+                {"path": str(s2), "role": "speech"},
+                {"path": str(music), "role": "music"},
+            ],
+            "ducking": {"enabled": True},
+            "output_path": str(out),
+        }
+    )
+
+    assert result.success is True, result.error
+    assert out.exists() and _has_audio(out)

tools/audio/audio_mixer.py

@@ -492,17 +492,22 @@ class AudioMixer(BaseTool):
         duck_enabled = ducking.get("enabled", True) if isinstance(ducking, dict) else bool(ducking)
 
         if duck_enabled and speech_tracks and music_tracks:
-            # Mix speech tracks together first
+            # Build ONE speech stream, then split it into two independent
+            # branches: one feeds the sidechain compressor as the ducking key,
+            # the other is mixed into the final output. A filtergraph label may
+            # only be consumed once, so reusing the same speech label for both
+            # the sidechain key and the output mix is invalid on stricter ffmpeg
+            # builds (e.g. the Linux ffmpeg on CI). asplit makes the fork explicit.
             speech_indices = list(range(len(speech_tracks)))
             speech_labels = "".join(f"[a{i}]" for i in speech_indices)
 
             if len(speech_tracks) > 1:
                 filter_parts.append(
-                    f"{speech_labels}amix=inputs={len(speech_tracks)}:duration=longest[speech_mix]"
+                    f"{speech_labels}amix=inputs={len(speech_tracks)}:duration=longest[speech_all]"
                 )
-                speech_out = "[speech_mix]"
             else:
-                speech_out = f"[a{speech_indices[0]}]"
+                filter_parts.append(f"[a{speech_indices[0]}]acopy[speech_all]")
+            filter_parts.append("[speech_all]asplit=2[speech_key][speech_out]")
 
             # Mix music tracks together
             music_start = len(speech_tracks)
@@ -517,42 +522,20 @@ class AudioMixer(BaseTool):
             else:
                 music_in = f"[a{music_indices[0]}]"
 
-            # Apply sidechain ducking
+            # Apply sidechain ducking — music is compressed, [speech_key] is the key
             duck_params = ducking if isinstance(ducking, dict) else {}
             attack = duck_params.get("attack_ms", 200) / 1000
             release = duck_params.get("release_ms", 500) / 1000
             music_vol = duck_params.get("music_volume_during_speech", 0.15)
 
             filter_parts.append(
-                f"{music_in}{speech_out}sidechaincompress="
+                f"{music_in}[speech_key]sidechaincompress="
                 f"threshold=0.02:ratio=9:attack={attack}:release={release}:"
                 f"level_sc=1:mix=0.9[ducked_music];"
                 f"[ducked_music]volume={music_vol * 3}[music_out]"
             )
 
-            # Duplicate speech for final mix (sidechain consumes it as key)
-            filter_parts.append(
-                f"{speech_out}acopy[speech_dup]" if speech_out.startswith("[a") else ""
-            )
-            # Re-mix speech path: we need speech audio in the output too
-            # Simpler approach: use amix on original speech and ducked music
-            # Reset: use a cleaner approach — amerge the speech mix and ducked music
-            # Actually, let's rebuild. The sidechain approach above uses speech as
-            # the key signal but doesn't consume it from the output chain.
-            # FFmpeg sidechaincompress: input 0 = audio to compress, input 1 = key signal
-            # So music is compressed, speech signal is the key. We need to mix them.
-            # Remove the last filter_part (the acopy that may be empty)
-            if filter_parts and filter_parts[-1] == "":
-                filter_parts.pop()
-
-            # Build speech mix for output separately
-            if len(speech_tracks) > 1:
-                # speech_mix already exists, make a copy for output
-                filter_parts.append(f"{speech_labels}amix=inputs={len(speech_tracks)}:duration=longest[speech_out]")
-            else:
-                filter_parts.append(f"[a{speech_indices[0]}]acopy[speech_out]")
-
-            # Final mix: speech_out + music_out
+            # Final mix: the other speech branch + ducked music
             mix_label = "[speech_out][music_out]amix=inputs=2:duration=longest[premix]"
 
             # Add SFX if present