Use Job infrastructure for Debug Replay (#23099)

* use ReplayState enum

* extract shared ffmpeg progress helper

* make start call non-blocking with worker thread

* expose replay state on status endpoint and return 202 from start

* cancel in-flight ffmpeg when stop is called during preparation

* add replay i18n strings for preparing and error states

* show status in replay UI

* navigate immediately on 202 from debug replay menus and dialog

* remove unused

* simplify to use Job infrastructure

* tests

* cleanup and tweaks

* fetch schema

* update api spec

* formatting

* fix e2e test

* mypy

* clean up

* formatting

* fix

* fix test

* don't try to show camera image until status reports ready

* simplify loading logic

* fix race in latest_frame on debug replay shutdown

* remove toast when successfully stopping

it gets hidden almost immediately

frigate/test/http_api/test_debug_replay_api.py

@@ -0,0 +1,123 @@
+"""Tests for /debug_replay API endpoints."""
+
+from unittest.mock import patch
+
+from frigate.models import Event, Recordings, ReviewSegment
+from frigate.test.http_api.base_http_test import AuthTestClient, BaseTestHttp
+
+
+class TestDebugReplayAPI(BaseTestHttp):
+    def setUp(self):
+        super().setUp([Event, Recordings, ReviewSegment])
+        self.app = self.create_app()
+
+    def test_start_returns_202_with_job_id(self):
+        # Stub the factory to skip validation/threading and just record the
+        # name on the manager the way the real factory's mark_starting would.
+        def fake_start(**kwargs):
+            kwargs["replay_manager"].mark_starting(
+                source_camera=kwargs["source_camera"],
+                replay_camera_name="_replay_front",
+                start_ts=kwargs["start_ts"],
+                end_ts=kwargs["end_ts"],
+            )
+            return "job-1234"
+
+        with patch(
+            "frigate.api.debug_replay.start_debug_replay_job",
+            side_effect=fake_start,
+        ):
+            with AuthTestClient(self.app) as client:
+                resp = client.post(
+                    "/debug_replay/start",
+                    json={
+                        "camera": "front",
+                        "start_time": 100,
+                        "end_time": 200,
+                    },
+                )
+
+        self.assertEqual(resp.status_code, 202)
+        body = resp.json()
+        self.assertTrue(body["success"])
+        self.assertEqual(body["job_id"], "job-1234")
+        self.assertEqual(body["replay_camera"], "_replay_front")
+
+    def test_start_returns_400_on_validation_error(self):
+        with patch(
+            "frigate.api.debug_replay.start_debug_replay_job",
+            side_effect=ValueError("Camera 'missing' not found"),
+        ):
+            with AuthTestClient(self.app) as client:
+                resp = client.post(
+                    "/debug_replay/start",
+                    json={
+                        "camera": "missing",
+                        "start_time": 100,
+                        "end_time": 200,
+                    },
+                )
+
+        self.assertEqual(resp.status_code, 400)
+        body = resp.json()
+        self.assertFalse(body["success"])
+        # Message is hard-coded so we don't echo exception text back to clients
+        # (CodeQL: information exposure through an exception).
+        self.assertEqual(body["message"], "Invalid debug replay parameters")
+
+    def test_start_returns_409_when_session_already_active(self):
+        with patch(
+            "frigate.api.debug_replay.start_debug_replay_job",
+            side_effect=RuntimeError("A replay session is already active"),
+        ):
+            with AuthTestClient(self.app) as client:
+                resp = client.post(
+                    "/debug_replay/start",
+                    json={
+                        "camera": "front",
+                        "start_time": 100,
+                        "end_time": 200,
+                    },
+                )
+
+        self.assertEqual(resp.status_code, 409)
+        body = resp.json()
+        self.assertFalse(body["success"])
+
+    def test_status_inactive_when_no_session(self):
+        with AuthTestClient(self.app) as client:
+            resp = client.get("/debug_replay/status")
+
+        self.assertEqual(resp.status_code, 200)
+        body = resp.json()
+        self.assertFalse(body["active"])
+        self.assertIsNone(body["replay_camera"])
+        self.assertIsNone(body["source_camera"])
+        self.assertIsNone(body["start_time"])
+        self.assertIsNone(body["end_time"])
+        self.assertFalse(body["live_ready"])
+        # Make sure deprecated fields are gone
+        self.assertNotIn("state", body)
+        self.assertNotIn("progress_percent", body)
+        self.assertNotIn("error_message", body)
+
+    def test_status_active_after_mark_starting(self):
+        manager = self.app.replay_manager
+        manager.mark_starting(
+            source_camera="front",
+            replay_camera_name="_replay_front",
+            start_ts=100.0,
+            end_ts=200.0,
+        )
+
+        with AuthTestClient(self.app) as client:
+            resp = client.get("/debug_replay/status")
+
+        self.assertEqual(resp.status_code, 200)
+        body = resp.json()
+        self.assertTrue(body["active"])
+        self.assertEqual(body["replay_camera"], "_replay_front")
+        self.assertEqual(body["source_camera"], "front")
+        self.assertEqual(body["start_time"], 100.0)
+        self.assertEqual(body["end_time"], 200.0)
+        self.assertFalse(body["live_ready"])

frigate/test/test_debug_replay.py

@@ -0,0 +1,242 @@
+"""Tests for the simplified DebugReplayManager.
+
+Startup orchestration lives in ``frigate.jobs.debug_replay`` (covered by
+``test_debug_replay_job``). The manager owns only session presence and
+cleanup.
+"""
+
+import unittest
+import unittest.mock
+from unittest.mock import MagicMock, patch
+
+
+class TestDebugReplayManagerSession(unittest.TestCase):
+    def test_inactive_by_default(self) -> None:
+        from frigate.debug_replay import DebugReplayManager
+
+        manager = DebugReplayManager()
+
+        self.assertFalse(manager.active)
+        self.assertIsNone(manager.replay_camera_name)
+        self.assertIsNone(manager.source_camera)
+        self.assertIsNone(manager.clip_path)
+        self.assertIsNone(manager.start_ts)
+        self.assertIsNone(manager.end_ts)
+
+    def test_mark_starting_sets_session_pointers_and_active(self) -> None:
+        from frigate.debug_replay import DebugReplayManager
+
+        manager = DebugReplayManager()
+
+        manager.mark_starting(
+            source_camera="front",
+            replay_camera_name="_replay_front",
+            start_ts=100.0,
+            end_ts=200.0,
+        )
+
+        self.assertTrue(manager.active)
+        self.assertEqual(manager.replay_camera_name, "_replay_front")
+        self.assertEqual(manager.source_camera, "front")
+        self.assertEqual(manager.start_ts, 100.0)
+        self.assertEqual(manager.end_ts, 200.0)
+        self.assertIsNone(manager.clip_path)
+
+    def test_mark_session_ready_sets_clip_path(self) -> None:
+        from frigate.debug_replay import DebugReplayManager
+
+        manager = DebugReplayManager()
+        manager.mark_starting("front", "_replay_front", 100.0, 200.0)
+
+        manager.mark_session_ready(clip_path="/tmp/replay/_replay_front.mp4")
+
+        self.assertEqual(manager.clip_path, "/tmp/replay/_replay_front.mp4")
+        self.assertTrue(manager.active)
+
+    def test_clear_session_resets_all_pointers(self) -> None:
+        from frigate.debug_replay import DebugReplayManager
+
+        manager = DebugReplayManager()
+        manager.mark_starting("front", "_replay_front", 100.0, 200.0)
+        manager.mark_session_ready("/tmp/replay/clip.mp4")
+
+        manager.clear_session()
+
+        self.assertFalse(manager.active)
+        self.assertIsNone(manager.replay_camera_name)
+        self.assertIsNone(manager.source_camera)
+        self.assertIsNone(manager.clip_path)
+        self.assertIsNone(manager.start_ts)
+        self.assertIsNone(manager.end_ts)
+
+
+class TestDebugReplayManagerStop(unittest.TestCase):
+    def test_stop_when_inactive_is_a_noop(self) -> None:
+        from frigate.debug_replay import DebugReplayManager
+
+        manager = DebugReplayManager()
+        frigate_config = MagicMock()
+        frigate_config.cameras = {}
+        publisher = MagicMock()
+
+        # Should not raise; should not publish any events.
+        manager.stop(frigate_config=frigate_config, config_publisher=publisher)
+
+        publisher.publish_update.assert_not_called()
+
+    def test_stop_publishes_remove_when_camera_was_published(self) -> None:
+        from frigate.config.camera.updater import CameraConfigUpdateEnum
+        from frigate.debug_replay import DebugReplayManager
+
+        manager = DebugReplayManager()
+        manager.mark_starting("front", "_replay_front", 100.0, 200.0)
+        manager.mark_session_ready("/tmp/replay/_replay_front.mp4")
+
+        camera_config = MagicMock()
+        frigate_config = MagicMock()
+        frigate_config.cameras = {"_replay_front": camera_config}
+        publisher = MagicMock()
+
+        with (
+            patch.object(manager, "_cleanup_db"),
+            patch.object(manager, "_cleanup_files"),
+            patch("frigate.debug_replay.cancel_debug_replay_job", return_value=False),
+        ):
+            manager.stop(frigate_config=frigate_config, config_publisher=publisher)
+
+        # One publish_update call with a remove topic.
+        self.assertEqual(publisher.publish_update.call_count, 1)
+        topic_arg = publisher.publish_update.call_args.args[0]
+        self.assertEqual(topic_arg.update_type, CameraConfigUpdateEnum.remove)
+        self.assertFalse(manager.active)
+
+    def test_stop_skips_remove_publish_when_camera_not_in_config(self) -> None:
+        """Cancellation during preparing_clip: no camera was published yet."""
+        from frigate.debug_replay import DebugReplayManager
+
+        manager = DebugReplayManager()
+        manager.mark_starting("front", "_replay_front", 100.0, 200.0)
+        # clip_path stays None because we cancelled before camera publish.
+
+        frigate_config = MagicMock()
+        frigate_config.cameras = {}  # _replay_front not present
+        publisher = MagicMock()
+
+        with (
+            patch.object(manager, "_cleanup_db"),
+            patch.object(manager, "_cleanup_files"),
+            patch("frigate.debug_replay.cancel_debug_replay_job", return_value=True),
+        ):
+            manager.stop(frigate_config=frigate_config, config_publisher=publisher)
+
+        publisher.publish_update.assert_not_called()
+        self.assertFalse(manager.active)
+
+    def test_stop_calls_cancel_debug_replay_job(self) -> None:
+        from frigate.debug_replay import DebugReplayManager
+
+        manager = DebugReplayManager()
+        manager.mark_starting("front", "_replay_front", 100.0, 200.0)
+
+        frigate_config = MagicMock()
+        frigate_config.cameras = {}
+        publisher = MagicMock()
+
+        with (
+            patch.object(manager, "_cleanup_db"),
+            patch.object(manager, "_cleanup_files"),
+            patch(
+                "frigate.debug_replay.cancel_debug_replay_job",
+                return_value=True,
+            ) as mock_cancel,
+        ):
+            manager.stop(frigate_config=frigate_config, config_publisher=publisher)
+
+        mock_cancel.assert_called_once()
+
+
+class TestDebugReplayManagerPublishCamera(unittest.TestCase):
+    def test_publish_camera_invokes_publisher_with_add_topic(self) -> None:
+        from frigate.config.camera.updater import CameraConfigUpdateEnum
+        from frigate.debug_replay import DebugReplayManager
+
+        manager = DebugReplayManager()
+
+        source_config = MagicMock()
+        new_camera_config = MagicMock()
+        frigate_config = MagicMock()
+        frigate_config.cameras = {"front": source_config}
+        publisher = MagicMock()
+
+        with (
+            patch.object(
+                manager,
+                "_build_camera_config_dict",
+                return_value={"enabled": True},
+            ),
+            patch("frigate.debug_replay.find_config_file", return_value="/cfg.yml"),
+            patch("frigate.debug_replay.YAML") as yaml_cls,
+            patch("frigate.debug_replay.FrigateConfig.parse_object") as parse_object,
+            patch("builtins.open", unittest.mock.mock_open(read_data="cameras:\n")),
+        ):
+            yaml_instance = yaml_cls.return_value
+            yaml_instance.load.return_value = {"cameras": {}}
+            parsed = MagicMock()
+            parsed.cameras = {"_replay_front": new_camera_config}
+            parse_object.return_value = parsed
+
+            manager.publish_camera(
+                source_camera="front",
+                replay_name="_replay_front",
+                clip_path="/tmp/clip.mp4",
+                frigate_config=frigate_config,
+                config_publisher=publisher,
+            )
+
+        # Camera registered into the live config dict
+        self.assertIn("_replay_front", frigate_config.cameras)
+        # Publisher invoked with an add topic
+        self.assertEqual(publisher.publish_update.call_count, 1)
+        topic_arg = publisher.publish_update.call_args.args[0]
+        self.assertEqual(topic_arg.update_type, CameraConfigUpdateEnum.add)
+
+    def test_publish_camera_wraps_parse_failure_in_runtime_error(self) -> None:
+        from frigate.debug_replay import DebugReplayManager
+
+        manager = DebugReplayManager()
+        frigate_config = MagicMock()
+        frigate_config.cameras = {"front": MagicMock()}
+        publisher = MagicMock()
+
+        with (
+            patch.object(
+                manager,
+                "_build_camera_config_dict",
+                return_value={"enabled": True},
+            ),
+            patch("frigate.debug_replay.find_config_file", return_value="/cfg.yml"),
+            patch("frigate.debug_replay.YAML") as yaml_cls,
+            patch(
+                "frigate.debug_replay.FrigateConfig.parse_object",
+                side_effect=ValueError("zone foo has invalid coordinates"),
+            ),
+            patch("builtins.open", unittest.mock.mock_open(read_data="cameras:\n")),
+        ):
+            yaml_cls.return_value.load.return_value = {"cameras": {}}
+
+            with self.assertRaises(RuntimeError) as ctx:
+                manager.publish_camera(
+                    source_camera="front",
+                    replay_name="_replay_front",
+                    clip_path="/tmp/clip.mp4",
+                    frigate_config=frigate_config,
+                    config_publisher=publisher,
+                )
+
+        self.assertIn("replay camera config", str(ctx.exception))
+        self.assertIn("invalid coordinates", str(ctx.exception))
+        publisher.publish_update.assert_not_called()
+
+
+if __name__ == "__main__":
+    unittest.main()

frigate/test/test_debug_replay_job.py

@@ -0,0 +1,460 @@
+"""Tests for the debug replay job runner and factory."""
+
+import threading
+import time
+import unittest
+import unittest.mock
+from unittest.mock import MagicMock, patch
+
+from frigate.debug_replay import DebugReplayManager
+from frigate.jobs.debug_replay import (
+    DebugReplayJob,
+    cancel_debug_replay_job,
+    get_active_runner,
+    start_debug_replay_job,
+)
+from frigate.jobs.export import JobStatePublisher
+from frigate.jobs.manager import _completed_jobs, _current_jobs
+from frigate.types import JobStatusTypesEnum
+
+
+def _reset_job_manager() -> None:
+    """Clear the global job manager state between tests."""
+    _current_jobs.clear()
+    _completed_jobs.clear()
+
+
+def _patch_publisher(test_case: unittest.TestCase) -> None:
+    """Replace JobStatePublisher.publish with a no-op to avoid hanging on IPC."""
+    publisher_patch = patch.object(
+        JobStatePublisher, "publish", lambda self, payload: None
+    )
+    publisher_patch.start()
+    test_case.addCleanup(publisher_patch.stop)
+
+
+class TestDebugReplayJob(unittest.TestCase):
+    def test_default_fields(self) -> None:
+        job = DebugReplayJob()
+
+        self.assertEqual(job.job_type, "debug_replay")
+        self.assertEqual(job.status, JobStatusTypesEnum.queued)
+        self.assertIsNone(job.current_step)
+        self.assertEqual(job.progress_percent, 0.0)
+
+    def test_to_dict_whitelist(self) -> None:
+        job = DebugReplayJob(
+            source_camera="front",
+            replay_camera_name="_replay_front",
+            start_ts=100.0,
+            end_ts=200.0,
+        )
+        job.current_step = "preparing_clip"
+        job.progress_percent = 42.5
+
+        payload = job.to_dict()
+
+        # Top-level matches the standard Job<TResults> shape.
+        for key in (
+            "id",
+            "job_type",
+            "status",
+            "start_time",
+            "end_time",
+            "error_message",
+            "results",
+        ):
+            self.assertIn(key, payload, f"missing top-level field: {key}")
+
+        results = payload["results"]
+        self.assertEqual(results["source_camera"], "front")
+        self.assertEqual(results["replay_camera_name"], "_replay_front")
+        self.assertEqual(results["current_step"], "preparing_clip")
+        self.assertEqual(results["progress_percent"], 42.5)
+        self.assertEqual(results["start_ts"], 100.0)
+        self.assertEqual(results["end_ts"], 200.0)
+
+
+class TestStartDebugReplayJob(unittest.TestCase):
+    def setUp(self) -> None:
+        _reset_job_manager()
+        _patch_publisher(self)
+        self.manager = DebugReplayManager()
+        self.frigate_config = MagicMock()
+        self.frigate_config.cameras = {"front": MagicMock()}
+        self.frigate_config.ffmpeg.ffmpeg_path = "/bin/true"
+        self.publisher = MagicMock()
+
+        self.recordings_qs = MagicMock()
+        self.recordings_qs.count.return_value = 1
+        self.recordings_qs.__iter__.return_value = iter([MagicMock(path="/tmp/r1.mp4")])
+
+    def tearDown(self) -> None:
+        runner = get_active_runner()
+        if runner is not None:
+            runner.cancel()
+            runner.join(timeout=2.0)
+        _reset_job_manager()
+
+    def test_rejects_unknown_camera(self) -> None:
+        with self.assertRaises(ValueError):
+            start_debug_replay_job(
+                source_camera="missing",
+                start_ts=100.0,
+                end_ts=200.0,
+                frigate_config=self.frigate_config,
+                config_publisher=self.publisher,
+                replay_manager=self.manager,
+            )
+
+    def test_rejects_invalid_time_range(self) -> None:
+        with self.assertRaises(ValueError):
+            start_debug_replay_job(
+                source_camera="front",
+                start_ts=200.0,
+                end_ts=100.0,
+                frigate_config=self.frigate_config,
+                config_publisher=self.publisher,
+                replay_manager=self.manager,
+            )
+
+    def test_rejects_when_no_recordings(self) -> None:
+        empty_qs = MagicMock()
+        empty_qs.count.return_value = 0
+        with patch("frigate.jobs.debug_replay.query_recordings", return_value=empty_qs):
+            with self.assertRaises(ValueError):
+                start_debug_replay_job(
+                    source_camera="front",
+                    start_ts=100.0,
+                    end_ts=200.0,
+                    frigate_config=self.frigate_config,
+                    config_publisher=self.publisher,
+                    replay_manager=self.manager,
+                )
+
+    def test_returns_job_id_and_marks_session_starting(self) -> None:
+        block = threading.Event()
+
+        def slow_helper(cmd, **kwargs):
+            block.wait(timeout=5)
+            return 0, ""
+
+        with (
+            patch(
+                "frigate.jobs.debug_replay.query_recordings",
+                return_value=self.recordings_qs,
+            ),
+            patch(
+                "frigate.jobs.debug_replay.run_ffmpeg_with_progress",
+                side_effect=slow_helper,
+            ),
+            patch.object(self.manager, "publish_camera"),
+            patch("os.path.exists", return_value=True),
+            patch("os.makedirs"),
+            patch("builtins.open", unittest.mock.mock_open()),
+        ):
+            job_id = start_debug_replay_job(
+                source_camera="front",
+                start_ts=100.0,
+                end_ts=200.0,
+                frigate_config=self.frigate_config,
+                config_publisher=self.publisher,
+                replay_manager=self.manager,
+            )
+
+            self.assertIsInstance(job_id, str)
+            self.assertTrue(self.manager.active)
+            self.assertEqual(self.manager.replay_camera_name, "_replay_front")
+            self.assertEqual(self.manager.source_camera, "front")
+
+            block.set()
+
+    def test_rejects_concurrent_calls(self) -> None:
+        block = threading.Event()
+
+        def slow_helper(cmd, **kwargs):
+            block.wait(timeout=5)
+            return 0, ""
+
+        with (
+            patch(
+                "frigate.jobs.debug_replay.query_recordings",
+                return_value=self.recordings_qs,
+            ),
+            patch(
+                "frigate.jobs.debug_replay.run_ffmpeg_with_progress",
+                side_effect=slow_helper,
+            ),
+            patch.object(self.manager, "publish_camera"),
+            patch("os.path.exists", return_value=True),
+            patch("os.makedirs"),
+            patch("builtins.open", unittest.mock.mock_open()),
+        ):
+            start_debug_replay_job(
+                source_camera="front",
+                start_ts=100.0,
+                end_ts=200.0,
+                frigate_config=self.frigate_config,
+                config_publisher=self.publisher,
+                replay_manager=self.manager,
+            )
+
+            with self.assertRaises(RuntimeError):
+                start_debug_replay_job(
+                    source_camera="front",
+                    start_ts=100.0,
+                    end_ts=200.0,
+                    frigate_config=self.frigate_config,
+                    config_publisher=self.publisher,
+                    replay_manager=self.manager,
+                )
+
+            block.set()
+
+
+class TestRunnerHappyPath(unittest.TestCase):
+    def setUp(self) -> None:
+        _reset_job_manager()
+        _patch_publisher(self)
+        self.manager = DebugReplayManager()
+        self.frigate_config = MagicMock()
+        self.frigate_config.cameras = {"front": MagicMock()}
+        self.frigate_config.ffmpeg.ffmpeg_path = "/bin/true"
+        self.publisher = MagicMock()
+
+        self.recordings_qs = MagicMock()
+        self.recordings_qs.count.return_value = 1
+        self.recordings_qs.__iter__.return_value = iter([MagicMock(path="/tmp/r1.mp4")])
+
+    def tearDown(self) -> None:
+        runner = get_active_runner()
+        if runner is not None:
+            runner.cancel()
+            runner.join(timeout=2.0)
+        _reset_job_manager()
+
+    def _wait_for(self, predicate, timeout: float = 5.0) -> bool:
+        deadline = time.time() + timeout
+        while time.time() < deadline:
+            if predicate():
+                return True
+            time.sleep(0.02)
+        return False
+
+    def test_progress_callback_updates_job_percent(self) -> None:
+        captured: list[float] = []
+
+        def fake_helper(cmd, *, on_progress=None, **kwargs):
+            on_progress(0.0)
+            on_progress(50.0)
+            on_progress(100.0)
+            return 0, ""
+
+        with (
+            patch(
+                "frigate.jobs.debug_replay.query_recordings",
+                return_value=self.recordings_qs,
+            ),
+            patch(
+                "frigate.jobs.debug_replay.run_ffmpeg_with_progress",
+                side_effect=fake_helper,
+            ),
+            patch.object(
+                self.manager,
+                "publish_camera",
+                side_effect=lambda *a, **kw: captured.append("published"),
+            ),
+            patch("os.path.exists", return_value=True),
+            patch("os.makedirs"),
+            patch("builtins.open", unittest.mock.mock_open()),
+        ):
+            start_debug_replay_job(
+                source_camera="front",
+                start_ts=100.0,
+                end_ts=200.0,
+                frigate_config=self.frigate_config,
+                config_publisher=self.publisher,
+                replay_manager=self.manager,
+            )
+
+            self.assertTrue(
+                self._wait_for(lambda: get_active_runner() is None),
+                "runner did not finish",
+            )
+
+        from frigate.jobs.manager import get_current_job
+
+        job = get_current_job("debug_replay")
+        self.assertIsNotNone(job)
+        self.assertEqual(job.status, JobStatusTypesEnum.success)
+        self.assertEqual(job.progress_percent, 100.0)
+        self.assertEqual(captured, ["published"])
+        # Manager should have been told the session is ready with the clip path.
+        self.assertIsNotNone(self.manager.clip_path)
+
+
+class TestRunnerFailurePath(unittest.TestCase):
+    def setUp(self) -> None:
+        _reset_job_manager()
+        _patch_publisher(self)
+        self.manager = DebugReplayManager()
+        self.frigate_config = MagicMock()
+        self.frigate_config.cameras = {"front": MagicMock()}
+        self.frigate_config.ffmpeg.ffmpeg_path = "/bin/true"
+        self.publisher = MagicMock()
+        self.recordings_qs = MagicMock()
+        self.recordings_qs.count.return_value = 1
+        self.recordings_qs.__iter__.return_value = iter([MagicMock(path="/tmp/r1.mp4")])
+
+    def tearDown(self) -> None:
+        runner = get_active_runner()
+        if runner is not None:
+            runner.cancel()
+            runner.join(timeout=2.0)
+        _reset_job_manager()
+
+    def _wait_for(self, predicate, timeout: float = 5.0) -> bool:
+        deadline = time.time() + timeout
+        while time.time() < deadline:
+            if predicate():
+                return True
+            time.sleep(0.02)
+        return False
+
+    def test_ffmpeg_failure_marks_job_failed_and_clears_session(self) -> None:
+        def failing_helper(cmd, **kwargs):
+            return 1, "ffmpeg exploded"
+
+        with (
+            patch(
+                "frigate.jobs.debug_replay.query_recordings",
+                return_value=self.recordings_qs,
+            ),
+            patch(
+                "frigate.jobs.debug_replay.run_ffmpeg_with_progress",
+                side_effect=failing_helper,
+            ),
+            patch("os.path.exists", return_value=True),
+            patch("os.makedirs"),
+            patch("os.remove"),
+            patch("builtins.open", unittest.mock.mock_open()),
+        ):
+            start_debug_replay_job(
+                source_camera="front",
+                start_ts=100.0,
+                end_ts=200.0,
+                frigate_config=self.frigate_config,
+                config_publisher=self.publisher,
+                replay_manager=self.manager,
+            )
+
+            self.assertTrue(
+                self._wait_for(lambda: get_active_runner() is None),
+                "runner did not finish",
+            )
+
+        from frigate.jobs.manager import get_current_job
+
+        job = get_current_job("debug_replay")
+        self.assertIsNotNone(job)
+        self.assertEqual(job.status, JobStatusTypesEnum.failed)
+        self.assertIsNotNone(job.error_message)
+        self.assertIn("ffmpeg", job.error_message.lower())
+        # Session cleared so a new /start is allowed
+        self.assertFalse(self.manager.active)
+
+
+class TestRunnerCancellation(unittest.TestCase):
+    def setUp(self) -> None:
+        _reset_job_manager()
+        _patch_publisher(self)
+        self.manager = DebugReplayManager()
+        self.frigate_config = MagicMock()
+        self.frigate_config.cameras = {"front": MagicMock()}
+        self.frigate_config.ffmpeg.ffmpeg_path = "/bin/true"
+        self.publisher = MagicMock()
+        self.recordings_qs = MagicMock()
+        self.recordings_qs.count.return_value = 1
+        self.recordings_qs.__iter__.return_value = iter([MagicMock(path="/tmp/r1.mp4")])
+
+    def tearDown(self) -> None:
+        runner = get_active_runner()
+        if runner is not None:
+            runner.cancel()
+            runner.join(timeout=2.0)
+        _reset_job_manager()
+
+    def _wait_for(self, predicate, timeout: float = 5.0) -> bool:
+        deadline = time.time() + timeout
+        while time.time() < deadline:
+            if predicate():
+                return True
+            time.sleep(0.02)
+        return False
+
+    def test_cancel_terminates_ffmpeg_and_marks_cancelled(self) -> None:
+        terminated = threading.Event()
+        fake_proc = MagicMock()
+        fake_proc.terminate = MagicMock(side_effect=lambda: terminated.set())
+
+        def fake_helper(cmd, *, process_started=None, **kwargs):
+            if process_started is not None:
+                process_started(fake_proc)
+            terminated.wait(timeout=5)
+            return -15, "killed"
+
+        with (
+            patch(
+                "frigate.jobs.debug_replay.query_recordings",
+                return_value=self.recordings_qs,
+            ),
+            patch(
+                "frigate.jobs.debug_replay.run_ffmpeg_with_progress",
+                side_effect=fake_helper,
+            ),
+            patch("os.path.exists", return_value=True),
+            patch("os.makedirs"),
+            patch("os.remove"),
+            patch("builtins.open", unittest.mock.mock_open()),
+        ):
+            start_debug_replay_job(
+                source_camera="front",
+                start_ts=100.0,
+                end_ts=200.0,
+                frigate_config=self.frigate_config,
+                config_publisher=self.publisher,
+                replay_manager=self.manager,
+            )
+
+            # Wait for the runner to register the active process.
+            self.assertTrue(
+                self._wait_for(
+                    lambda: (
+                        get_active_runner() is not None
+                        and get_active_runner()._active_process is fake_proc
+                    )
+                )
+            )
+
+            cancelled = cancel_debug_replay_job()
+            self.assertTrue(cancelled)
+            self.assertTrue(fake_proc.terminate.called)
+
+            self.assertTrue(
+                self._wait_for(lambda: get_active_runner() is None),
+                "runner did not finish",
+            )
+
+        from frigate.jobs.manager import get_current_job
+
+        job = get_current_job("debug_replay")
+        self.assertEqual(job.status, JobStatusTypesEnum.cancelled)
+        # Runner must not clear the manager session on cancellation —
+        # that belongs to the caller of cancel_debug_replay_job (stop()).
+        # If the runner cleared it, stop() would log "no active session"
+        # and skip its cleanup_db / cleanup_files calls.
+        self.assertTrue(self.manager.active)
+
+
+if __name__ == "__main__":
+    unittest.main()

frigate/test/test_export_progress.py

@@ -14,6 +14,7 @@ from frigate.jobs.export import (
 )
 from frigate.record.export import PlaybackSourceEnum, RecordingExporter
 from frigate.types import JobStatusTypesEnum
+from frigate.util.ffmpeg import inject_progress_flags
 
 
 def _make_exporter(
@@ -118,10 +119,9 @@ class TestExpectedOutputDuration(unittest.TestCase):
 
 class TestProgressFlagInjection(unittest.TestCase):
     def test_inserts_before_output_path(self) -> None:
-        exporter = _make_exporter()
         cmd = ["ffmpeg", "-i", "input.m3u8", "-c", "copy", "/tmp/output.mp4"]
 
-        result = exporter._inject_progress_flags(cmd)
+        result = inject_progress_flags(cmd)
 
         assert result == [
             "ffmpeg",
@@ -136,8 +136,7 @@ class TestProgressFlagInjection(unittest.TestCase):
         ]
 
     def test_handles_empty_cmd(self) -> None:
-        exporter = _make_exporter()
-        assert exporter._inject_progress_flags([]) == []
+        assert inject_progress_flags([]) == []
 
 
 class TestFfmpegProgressParsing(unittest.TestCase):
@@ -167,7 +166,7 @@ class TestFfmpegProgressParsing(unittest.TestCase):
         fake_proc.returncode = 0
         fake_proc.wait = MagicMock(return_value=0)
 
-        with patch("frigate.record.export.sp.Popen", return_value=fake_proc):
+        with patch("frigate.util.ffmpeg.sp.Popen", return_value=fake_proc):
             returncode, _stderr = exporter._run_ffmpeg_with_progress(
                 ["ffmpeg", "-i", "x.m3u8", "/tmp/out.mp4"], "playlist", step="encoding"
             )

frigate/test/test_ffmpeg_progress.py

@@ -0,0 +1,111 @@
+"""Tests for the shared ffmpeg progress helper."""
+
+import unittest
+from unittest.mock import MagicMock, patch
+
+from frigate.util.ffmpeg import inject_progress_flags, run_ffmpeg_with_progress
+
+
+class TestInjectProgressFlags(unittest.TestCase):
+    def test_inserts_flags_before_output_path(self):
+        cmd = ["ffmpeg", "-i", "in.mp4", "-c", "copy", "out.mp4"]
+        result = inject_progress_flags(cmd)
+        self.assertEqual(
+            result,
+            [
+                "ffmpeg",
+                "-i",
+                "in.mp4",
+                "-c",
+                "copy",
+                "-progress",
+                "pipe:2",
+                "-nostats",
+                "out.mp4",
+            ],
+        )
+
+    def test_empty_cmd_returns_empty(self):
+        self.assertEqual(inject_progress_flags([]), [])
+
+
+class TestRunFfmpegWithProgress(unittest.TestCase):
+    def _make_fake_proc(self, stderr_lines, returncode=0):
+        proc = MagicMock()
+        proc.stderr = iter(stderr_lines)
+        proc.stdin = MagicMock()
+        proc.returncode = returncode
+        proc.wait = MagicMock()
+        return proc
+
+    def test_emits_percent_from_out_time_us_lines(self):
+        captured: list[float] = []
+
+        def on_progress(percent: float) -> None:
+            captured.append(percent)
+
+        stderr_lines = [
+            "out_time_us=1000000\n",
+            "out_time_us=5000000\n",
+            "progress=end\n",
+        ]
+        proc = self._make_fake_proc(stderr_lines)
+        proc.stderr = MagicMock()
+        proc.stderr.__iter__ = lambda self: iter(stderr_lines)
+        proc.stderr.read = MagicMock(return_value="")
+
+        with patch("subprocess.Popen", return_value=proc):
+            returncode, _stderr = run_ffmpeg_with_progress(
+                ["ffmpeg", "-i", "in", "out"],
+                expected_duration_seconds=10.0,
+                on_progress=on_progress,
+                use_low_priority=False,
+            )
+
+        self.assertEqual(returncode, 0)
+        self.assertEqual(len(captured), 4)  # initial 0.0 + two parsed + final 100.0
+        self.assertAlmostEqual(captured[0], 0.0)
+        self.assertAlmostEqual(captured[1], 10.0)
+        self.assertAlmostEqual(captured[2], 50.0)
+        self.assertAlmostEqual(captured[3], 100.0)
+
+    def test_passes_started_process_to_callback(self):
+        proc = self._make_fake_proc([])
+        proc.stderr = MagicMock()
+        proc.stderr.__iter__ = lambda self: iter([])
+        proc.stderr.read = MagicMock(return_value="")
+
+        seen: list = []
+
+        with patch("subprocess.Popen", return_value=proc):
+            run_ffmpeg_with_progress(
+                ["ffmpeg", "out"],
+                expected_duration_seconds=1.0,
+                process_started=lambda p: seen.append(p),
+                use_low_priority=False,
+            )
+
+        self.assertEqual(seen, [proc])
+
+    def test_clamps_percent_to_0_100(self):
+        captured: list[float] = []
+
+        def on_progress(percent: float) -> None:
+            captured.append(percent)
+
+        stderr_lines = ["out_time_us=999999999999\n"]
+        proc = self._make_fake_proc(stderr_lines)
+        proc.stderr = MagicMock()
+        proc.stderr.__iter__ = lambda self: iter(stderr_lines)
+        proc.stderr.read = MagicMock(return_value="")
+
+        with patch("subprocess.Popen", return_value=proc):
+            run_ffmpeg_with_progress(
+                ["ffmpeg", "out"],
+                expected_duration_seconds=10.0,
+                on_progress=on_progress,
+                use_low_priority=False,
+            )
+
+        # initial 0.0 then a clamped reading
+        self.assertEqual(captured[-1], 100.0)