refactor: govern background job lifecycle

backend/jobs/runner.py

@@ -18,6 +18,9 @@ class InProcessJobRunner:
         self.repository = repository
         self._locks: dict[str, threading.Lock] = {}
         self._locks_guard = threading.Lock()
+        self._scheduler_guard = threading.Lock()
+        self._scheduler_stop = threading.Event()
+        self._scheduler_thread: threading.Thread | None = None
 
     def submit(
         self, job_id: str, idempotency_key: str, action: JobAction,
@@ -52,27 +55,42 @@ class InProcessJobRunner:
         return True
 
     def start_scheduler(
-        self, callback: Callable[[], None], stop_event: threading.Event,
-        interval_seconds: float, initial_delay_seconds: float = 0,
+        self, callback: Callable[[], None], interval_seconds: float,
+        initial_delay_seconds: float = 0,
     ) -> threading.Thread:
-        def schedule_loop() -> None:
-            if stop_event.wait(initial_delay_seconds):
-                return
-            while not stop_event.is_set():
-                try:
-                    callback()
-                except Exception:
-                    # Submitted jobs persist their own failures; the scheduler must stay alive.
-                    pass
-                stop_event.wait(interval_seconds)
+        with self._scheduler_guard:
+            current = self._scheduler_thread
+            if current is not None and current.is_alive():
+                return current
+            self._scheduler_stop.clear()
 
-        thread = threading.Thread(
-            target=schedule_loop,
-            name="background-job-scheduler",
-            daemon=True,
-        )
-        thread.start()
-        return thread
+            def schedule_loop() -> None:
+                if self._scheduler_stop.wait(initial_delay_seconds):
+                    return
+                while not self._scheduler_stop.is_set():
+                    try:
+                        callback()
+                    except Exception:
+                        # Submitted jobs persist failures; the scheduler must stay alive.
+                        pass
+                    self._scheduler_stop.wait(interval_seconds)
+
+            thread = threading.Thread(
+                target=schedule_loop,
+                name="background-job-scheduler",
+                daemon=True,
+            )
+            self._scheduler_thread = thread
+            thread.start()
+            return thread
+
+    def stop_scheduler(self, timeout_seconds: float = 5) -> bool:
+        with self._scheduler_guard:
+            thread = self._scheduler_thread
+            self._scheduler_stop.set()
+            if thread is not None and thread is not threading.current_thread():
+                thread.join(max(0, timeout_seconds))
+            return thread is None or not thread.is_alive()
 
     def wait_for_idle(self, timeout_seconds: float = 5) -> bool:
         deadline = time.monotonic() + max(0, timeout_seconds)