fix(audio): schedule delayed track fades correctly

tests/tools/test_audio_mixer_track_fades.py

@@ -0,0 +1,95 @@
+"""Regression tests for delayed per-track fades in ``audio_mixer``."""
+
+import shutil
+import subprocess
+import sys
+from pathlib import Path
+from types import SimpleNamespace
+
+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
+
+
+@pytest.mark.parametrize("operation", ["mix", "full_mix"])
+def test_delayed_track_fades_use_the_source_timeline(tmp_path, monkeypatch, operation):
+    """Fade source audio before delay, and start fade-out at its real end."""
+    tone = tmp_path / "tone.wav"
+    tone.write_bytes(b"stub")
+    commands = []
+
+    def fake_run(self, cmd, **kwargs):
+        commands.append(list(cmd))
+        if cmd[0] == "ffprobe":
+            return SimpleNamespace(stdout="2.0\n", stderr="")
+        return SimpleNamespace(stdout="", stderr="")
+
+    monkeypatch.setattr(AudioMixer, "run_command", fake_run)
+    result = AudioMixer().execute(
+        {
+            "operation": operation,
+            "tracks": [{
+                "path": str(tone),
+                "role": "speech",
+                "start_seconds": 1,
+                "fade_in_seconds": 0.25,
+                "fade_out_seconds": 0.5,
+            }],
+            "ducking": {"enabled": False},
+            "normalize": False,
+            "output_path": str(tmp_path / "out.wav"),
+        }
+    )
+
+    assert result.success, result.error
+    ffmpeg_cmd = next(cmd for cmd in commands if cmd[0] == "ffmpeg")
+    filter_graph = ffmpeg_cmd[ffmpeg_cmd.index("-filter_complex") + 1]
+    assert (
+        "afade=t=in:d=0.25,afade=t=out:st=1.5:d=0.5,adelay=1000|1000"
+        in filter_graph
+    )
+
+
+@pytest.mark.skipif(shutil.which("ffmpeg") is None, reason="ffmpeg required")
+def test_delayed_track_with_fade_out_remains_audible_after_its_start(tmp_path):
+    """End-to-end regression for #347: delayed audio must not be silenced."""
+    tone = tmp_path / "tone.wav"
+    output = tmp_path / "mixed.wav"
+    subprocess.run(
+        ["ffmpeg", "-y", "-f", "lavfi", "-i", "sine=frequency=440:duration=2", str(tone)],
+        capture_output=True,
+        check=True,
+        timeout=30,
+    )
+
+    result = AudioMixer().execute(
+        {
+            "operation": "mix",
+            "tracks": [{
+                "path": str(tone),
+                "role": "music",
+                "start_seconds": 1,
+                "fade_out_seconds": 0.5,
+            }],
+            "normalize": False,
+            "output_path": str(output),
+        }
+    )
+    assert result.success, result.error
+
+    measured = subprocess.run(
+        [
+            "ffmpeg", "-ss", "1.25", "-t", "0.25", "-i", str(output),
+            "-vn", "-af", "volumedetect", "-f", "null", "-",
+        ],
+        capture_output=True,
+        text=True,
+        check=True,
+        timeout=30,
+    )
+    mean_line = next(line for line in measured.stderr.splitlines() if "mean_volume" in line)
+    mean_db = float(mean_line.split("mean_volume:")[1].strip().split(" ")[0])
+    assert mean_db > -60, f"delayed tone was unexpectedly silent: {mean_db} dB"

tools/audio/audio_mixer.py

@@ -213,6 +213,39 @@ class AudioMixer(BaseTool):
         target = max(-40.0, min(0.0, target))
         return f"[{in_label}]loudnorm=I={target}:LRA=11:TP=-1.5[{out_label}]"
 
+    def _track_filters(self, track: dict[str, Any]) -> list[str]:
+        """Build per-track filters on the source timeline before scheduling it.
+
+        ``afade=t=out`` defaults to ``st=0``. Applying it after ``adelay``
+        therefore fades the delay silence instead of the source audio, leaving
+        a delayed track silent by the time it starts. Fade source samples first
+        and add the timeline delay last so both fades follow the track itself.
+        """
+        filters = []
+        volume = track.get("volume", 1.0)
+        delay_ms = int(track.get("start_seconds", 0) * 1000)
+        fade_in = track.get("fade_in_seconds", 0)
+        fade_out = track.get("fade_out_seconds", 0)
+
+        if volume != 1.0:
+            filters.append(f"volume={volume}")
+        if fade_in > 0:
+            filters.append(f"afade=t=in:d={fade_in}")
+        if fade_out > 0:
+            duration_cmd = [
+                "ffprobe", "-v", "error",
+                "-show_entries", "format=duration",
+                "-of", "csv=p=0",
+                track["path"],
+            ]
+            duration = float(self.run_command(duration_cmd).stdout.strip().split("\n")[0])
+            fade_start = max(0.0, duration - float(fade_out))
+            filters.append(f"afade=t=out:st={fade_start}:d={fade_out}")
+        if delay_ms > 0:
+            filters.append(f"adelay={delay_ms}|{delay_ms}")
+
+        return filters
+
     def execute(self, inputs: dict[str, Any]) -> ToolResult:
         operation = inputs["operation"]
         start = time.time()
@@ -256,20 +289,7 @@ class AudioMixer(BaseTool):
 
         for i, track in enumerate(tracks):
             input_args.extend(["-i", track["path"]])
-            volume = track.get("volume", 1.0)
-            delay_ms = int(track.get("start_seconds", 0) * 1000)
-            fade_in = track.get("fade_in_seconds", 0)
-            fade_out = track.get("fade_out_seconds", 0)
-
-            filters = []
-            if volume != 1.0:
-                filters.append(f"volume={volume}")
-            if delay_ms > 0:
-                filters.append(f"adelay={delay_ms}|{delay_ms}")
-            if fade_in > 0:
-                filters.append(f"afade=t=in:d={fade_in}")
-            if fade_out > 0:
-                filters.append(f"afade=t=out:d={fade_out}")
+            filters = self._track_filters(track)
 
             if filters:
                 filter_chain = ",".join(filters)
@@ -500,20 +520,7 @@ class AudioMixer(BaseTool):
 
         for i, track in enumerate(all_tracks):
             input_args.extend(["-i", track["path"]])
-            volume = track.get("volume", 1.0)
-            delay_ms = int(track.get("start_seconds", 0) * 1000)
-            fade_in = track.get("fade_in_seconds", 0)
-            fade_out = track.get("fade_out_seconds", 0)
-
-            filters = []
-            if volume != 1.0:
-                filters.append(f"volume={volume}")
-            if delay_ms > 0:
-                filters.append(f"adelay={delay_ms}|{delay_ms}")
-            if fade_in > 0:
-                filters.append(f"afade=t=in:d={fade_in}")
-            if fade_out > 0:
-                filters.append(f"afade=t=out:d={fade_out}")
+            filters = self._track_filters(track)
 
             if filters:
                 filter_chain = ",".join(filters)