From a1612b9da29c8ce71849a34772100c333d30b102 Mon Sep 17 00:00:00 2001
From: CI <ci@example.com>
Date: Wed, 16 Sep 2026 09:15:52 +0100
Subject: [PATCH 1/5] Run ML training off the event loop and let a shutdown
 abandon it

Fixes #5075
---
 apps/predbat/load_ml_component.py  |  91 +++++++------
 apps/predbat/load_predictor.py     |  61 ++++++++-
 apps/predbat/tests/test_load_ml.py | 207 ++++++++++++++++++++++++-----
 3 files changed, 285 insertions(+), 74 deletions(-)

diff --git a/apps/predbat/load_ml_component.py b/apps/predbat/load_ml_component.py
index a84d0133f..0756f4653 100644
--- a/apps/predbat/load_ml_component.py
+++ b/apps/predbat/load_ml_component.py
@@ -777,6 +777,12 @@ async def run(self, seconds, first):
                     self.log("ML Component: Doing training...")
                     await self._do_training(is_initial)
 
+                    if self.api_stop:
+                        # Training abandoned for shutdown - a re-fetch and a prediction cycle here
+                        # would just add their own delay to the stop we are already answering
+                        self.log("ML Component: Stopping, skipping the post-training prediction cycle")
+                        return True
+
                     # Training can run for many minutes, during which the data fetched above
                     # goes stale - both the lookback window feeding the prediction and the
                     # load_minutes_now baseline still refer to the pre-training time. Re-fetch
@@ -945,6 +951,10 @@ def array_to_dict(arr):
             self.export_rates_data = None
             self.data_ready = False
 
+    def training_stop_requested(self):
+        """Return True when the component is stopping, so an in-flight training run abandons itself."""
+        return self.api_stop
+
     async def _do_training(self, is_initial):
         """
         Perform model training.
@@ -953,8 +963,9 @@ async def _do_training(self, is_initial):
             is_initial: True for full training, False for fine-tuning
         """
         # Snapshot data under the lock so we can release it before the CPU-bound
-        # training call.  predictor.train() can take 30-120 seconds on slow hardware
-        # and must NOT run while holding data_lock or the asyncio event loop will freeze.
+        # training call.  A curriculum run is many passes of many epochs and routinely takes
+        # minutes, so it must NOT run while holding data_lock, and must not run on the event
+        # loop either - see the to_thread call below.
         async with self.data_lock:
             if not self.load_data:
                 self.log("Warn: ML Component: No data for training")
@@ -974,48 +985,46 @@ async def _do_training(self, is_initial):
             time_decay = min(self.ml_time_decay_days, self.load_data_age_days)
             holdout_hours = self.ml_validation_holdout_hours
             patience = self.ml_patience_initial if is_initial else self.ml_patience_update
-            max_intermediate_passes = self.ml_curriculum_max_passes
             window_days = self.ml_curriculum_window_days
-            step_days = self.ml_curriculum_step_days
-        # Lock released - event loop is free during training
-
-        try:
+            # The initial curriculum walks the window out in fixed wider steps with a capped pass
+            # count rather than following the configured fine-tune sizing, so the first model is
+            # built from the oldest data forward without the run growing unbounded with history.
             if is_initial:
-                # Curriculum: progressively expand the training window from oldest week
-                # forward so the model learns gradually from historical structure.
-                val_mae = self.predictor.train_curriculum(
-                    load_data_snap,
-                    now_utc_snap,
-                    pv_minutes=pv_data_snap,
-                    temp_minutes=temp_data_snap,
-                    import_rates=import_rates_snap,
-                    export_rates=export_rates_snap,
-                    epochs=epochs,
-                    time_decay_days=time_decay,
-                    validation_holdout_hours=holdout_hours,
-                    patience=patience,
-                    curriculum_window_days=window_days,
-                    curriculum_step_days=5,
-                    max_intermediate_passes=8,
-                    progress_callback=self.update_success_timestamp,
-                )
+                step_days = 5
+                max_intermediate_passes = 8
             else:
-                val_mae = self.predictor.train_curriculum(
-                    load_data_snap,
-                    now_utc_snap,
-                    pv_minutes=pv_data_snap,
-                    temp_minutes=temp_data_snap,
-                    import_rates=import_rates_snap,
-                    export_rates=export_rates_snap,
-                    epochs=epochs,
-                    time_decay_days=time_decay,
-                    validation_holdout_hours=holdout_hours,
-                    patience=patience,
-                    curriculum_window_days=window_days,
-                    curriculum_step_days=step_days,
-                    max_intermediate_passes=max_intermediate_passes,
-                    progress_callback=self.update_success_timestamp,
-                )
+                step_days = self.ml_curriculum_step_days
+                max_intermediate_passes = self.ml_curriculum_max_passes
+        # Lock released
+
+        try:
+            # Run the curriculum on a worker thread. It is synchronous, CPU-bound and minutes long,
+            # so calling it directly would pin this component's event loop for its whole duration:
+            # api_stop would go unnoticed, the run_timeout watchdog could never fire, and a shutdown
+            # landing mid-training would sit waiting on this component's thread (#5075).
+            val_mae = await asyncio.to_thread(
+                self.predictor.train_curriculum,
+                load_data_snap,
+                now_utc_snap,
+                pv_minutes=pv_data_snap,
+                temp_minutes=temp_data_snap,
+                import_rates=import_rates_snap,
+                export_rates=export_rates_snap,
+                epochs=epochs,
+                time_decay_days=time_decay,
+                validation_holdout_hours=holdout_hours,
+                patience=patience,
+                curriculum_window_days=window_days,
+                curriculum_step_days=step_days,
+                max_intermediate_passes=max_intermediate_passes,
+                progress_callback=self.update_success_timestamp,
+                stop_callback=self.training_stop_requested,
+            )
+
+            if val_mae is None and self.training_stop_requested():
+                # Abandoned on purpose, not a failure - and nothing to save, the model is partial
+                self.log("ML Component: Training abandoned because the component is stopping")
+                return
 
             if val_mae is not None:
                 self.last_train_time = datetime.now(timezone.utc)
diff --git a/apps/predbat/load_predictor.py b/apps/predbat/load_predictor.py
index 1ed3dd0dd..a89196337 100644
--- a/apps/predbat/load_predictor.py
+++ b/apps/predbat/load_predictor.py
@@ -1170,6 +1170,16 @@ def _clip_predictions(self, predictions, lookback_buffer=None):
 
         return predictions
 
+    def _should_stop(self, stop_callback):
+        """Return True when the caller has asked training to abort, treating a raising hook as no."""
+        if not stop_callback:
+            return False
+        try:
+            return bool(stop_callback())
+        except Exception as e:
+            self.log("Warn: ML Predictor: stop callback raised {}, continuing training".format(e))
+            return False
+
     def train(
         self,
         load_minutes,
@@ -1188,6 +1198,7 @@ def train(
         ema_smoothing_alpha=0.3,
         huber_delta=1.35,
         progress_callback=None,
+        stop_callback=None,
     ):
         """
         Train or fine-tune the model.
@@ -1215,9 +1226,13 @@ def train(
             progress_callback: Called with no arguments once per epoch, so a caller whose liveness is
                                judged on how recently it reported success can stay alive through a
                                training run longer than that window. Exceptions are logged and swallowed
+            stop_callback: Called with no arguments at the top of every epoch; returning True abandons
+                           the run. Distinct from progress_callback precisely because that hook must
+                           never be able to abort a run - this one is the caller explicitly asking to.
+                           Exceptions are logged and treated as "do not stop"
 
         Returns:
-            Validation MAE or None if training failed
+            Validation MAE, or None if training failed or was aborted via stop_callback
         """
         self.log("ML Predictor: Starting {} training with {} epochs".format("initial" if is_initial else "fine-tune", epochs))
 
@@ -1309,7 +1324,16 @@ def train(
         # EMA accumulator for early-stopping metric (seeds on first epoch)
         ema_combined = None
 
+        aborted = False
         for epoch in range(epochs):
+            # Abort point. A single pass is minutes of uninterruptible CPU work, so without a check
+            # here a shutdown has to wait the whole run out (#5075). Checked before the epoch's work
+            # rather than after so the caller stops paying for an epoch it has already given up on.
+            if self._should_stop(stop_callback):
+                self.log("ML Predictor: Training aborted at epoch {}/{} - stop requested".format(epoch + 1, epochs))
+                aborted = True
+                break
+
             # Cosine LR decay: lr_t decays from lr_max at epoch 0 to lr_min at the final epoch
             if lr_decay == "cosine":
                 lr_t = lr_min + 0.5 * (lr_max - lr_min) * (1.0 + np.cos(np.pi * epoch / max(epochs - 1, 1)))
@@ -1418,6 +1442,12 @@ def train(
                 )
             )
 
+        if aborted:
+            # Weights are restored above so the in-memory model is left consistent, but the run did
+            # not finish: report failure so the caller neither publishes nor saves a partial model,
+            # and skip the AR rollout diagnostic, which is itself minutes of the work we just abandoned.
+            return None
+
         self.training_timestamp = datetime.now(timezone.utc)
         self.validation_mae = best_val_loss
         self.validation_bias = float(best_val_bias)
@@ -1493,6 +1523,7 @@ def train_curriculum(
         max_intermediate_passes=0,
         huber_delta=1.35,
         progress_callback=None,
+        stop_callback=None,
     ):
         """
                 Train using curriculum learning: progressively expand the training window
@@ -1526,6 +1557,8 @@ def train_curriculum(
                     progress_callback: Called with no arguments once per epoch across every pass, so a
                                        run that lasts longer than the caller's liveness timeout can
                                        still report itself alive
+                    stop_callback: Called with no arguments before each pass and, via train(), before each
+                                   epoch; returning True abandons the whole curriculum and returns None
                     curriculum_window_days: Initial training window size in days (default 7)
                     curriculum_step_days: Days added per subsequent pass (default 7)
                     max_intermediate_passes: Maximum number of intermediate passes to run;
@@ -1535,8 +1568,13 @@ def train_curriculum(
                     huber_delta: Huber loss transition point passed through to each train() call
 
                 Returns:
-                    Validation MAE from the final pass, or None if all passes failed.
+                    Validation MAE from the final pass, or None if all passes failed or the run was
+                    aborted via stop_callback.
         """
+        if self._should_stop(stop_callback):
+            self.log("ML Predictor: Curriculum training not started - stop requested")
+            return None
+
         # Build list of positive minute keys to find total history span
         hist_minutes = [k for k in load_minutes if isinstance(k, int) and k > 0]
         if not hist_minutes:
@@ -1581,6 +1619,7 @@ def train_curriculum(
                 norm_ema_alpha=norm_ema_alpha,
                 huber_delta=huber_delta,
                 progress_callback=progress_callback,
+                stop_callback=stop_callback,
             )
 
         total_passes = len(window_sizes) + 1  # intermediate passes + final full pass
@@ -1588,6 +1627,10 @@ def train_curriculum(
 
         val_mae = None
         for pass_idx, window in enumerate(window_sizes):
+            if self._should_stop(stop_callback):
+                self.log("ML Predictor: Curriculum training aborted before pass {}/{} - stop requested".format(pass_idx + 1, total_passes))
+                return None
+
             # Slice data to the oldest 'window' minutes.
             # start_minute is the more-recent edge; after re-indexing it becomes key 0.
             # We pass slice_now = now_utc - start_minute to train() so that the
@@ -1618,8 +1661,13 @@ def train_curriculum(
                 norm_ema_alpha=norm_ema_alpha,
                 huber_delta=huber_delta,
                 progress_callback=progress_callback,
+                stop_callback=stop_callback,
             )
 
+            if pass_mae is None and self._should_stop(stop_callback):
+                self.log("ML Predictor: Curriculum training aborted during pass {}/{} - stop requested".format(pass_idx + 1, total_passes))
+                return None
+
             if pass_mae is None:
                 self.log("Warn: ML Predictor: Curriculum pass {}/{} failed (insufficient data or training error) - skipping".format(pass_idx + 1, total_passes))
             else:
@@ -1627,6 +1675,10 @@ def train_curriculum(
                 self.log("ML Predictor: Curriculum pass {}/{} complete, val_mae={:.4f} kWh".format(pass_idx + 1, total_passes, val_mae))
 
         # Final pass: full dataset, standard holdout window
+        if self._should_stop(stop_callback):
+            self.log("ML Predictor: Curriculum training aborted before final pass {}/{} - stop requested".format(total_passes, total_passes))
+            return None
+
         self.log("ML Predictor: Curriculum final pass {}/{}: full dataset ({:.1f} days)".format(total_passes, total_passes, max_minute / day_minutes))
         final_mae = self.train(
             load_minutes,
@@ -1643,8 +1695,13 @@ def train_curriculum(
             norm_ema_alpha=norm_ema_alpha,
             huber_delta=huber_delta,
             progress_callback=progress_callback,
+            stop_callback=stop_callback,
         )
 
+        if final_mae is None and self._should_stop(stop_callback):
+            self.log("ML Predictor: Curriculum training aborted during final pass {}/{} - stop requested".format(total_passes, total_passes))
+            return None
+
         if final_mae is not None:
             val_mae = final_mae
 
diff --git a/apps/predbat/tests/test_load_ml.py b/apps/predbat/tests/test_load_ml.py
index dda419344..26944e2d8 100644
--- a/apps/predbat/tests/test_load_ml.py
+++ b/apps/predbat/tests/test_load_ml.py
@@ -56,6 +56,9 @@ def test_load_ml(my_predbat=None):
         ("curriculum_training", _test_curriculum_training, "Curriculum training with progressive window expansion"),
         ("training_progress_callback", _test_training_progress_callback, "Training reports liveness through progress_callback"),
         ("component_alive_during_training", _test_component_marks_itself_alive_during_training, "Component keeps its success timestamp fresh while training"),
+        ("training_stop_callback", _test_training_stops_on_stop_callback, "Training abandons a run when the stop hook trips"),
+        ("training_off_event_loop", _test_component_training_does_not_block_event_loop, "Component training leaves its event loop free to run"),
+        ("training_abandoned_on_stop", _test_component_training_abandons_run_on_stop, "Component abandons an in-flight training run when stopping"),
         ("prediction", _test_prediction, "End-to-end prediction"),
         ("prediction_with_pv", _test_prediction_with_pv, "Prediction with PV forecast data"),
         ("prediction_with_temp", _test_prediction_with_temp, "Prediction with temperature forecast data"),
@@ -3374,21 +3377,12 @@ def bad_callback():
     return True
 
 
-def _test_component_marks_itself_alive_during_training():
-    """_do_training must hand the trainer the component's own timestamp updater.
-
-    Driven through the real _do_training rather than by calling train_curriculum directly - a test
-    that passes the callback itself proves only that the test passes it, and would keep passing if
-    load_ml_component stopped wiring it up. That regression is invisible until a training run
-    silently crosses the liveness window again hours later.
-    """
+def _make_training_component(predictor, now_utc):
+    """Build a bare LoadMLComponent wired up with just the attributes _do_training reads."""
     import asyncio
 
     from load_ml_component import LoadMLComponent
 
-    now_utc = datetime.now(timezone.utc)
-    recorded = {}
-
     class FakeBase:
         """Minimal stand-in for the PredBat instance the component reads through."""
 
@@ -3400,27 +3394,12 @@ def log(self, msg, **kwargs):
             """Swallow component logging."""
             return None
 
-    class FakePredictor:
-        """Captures the kwargs the component passes, and pulses the callback like a real run."""
-
-        def train_curriculum(self, *args, **kwargs):
-            """Record every call's progress callback and fire it once, as an epoch would.
-
-            Every call, not just the last: with is_initial the component runs the initial curriculum
-            and then a fine-tune pass, so keeping only the most recent kwargs would let an unwired
-            first call hide behind a correctly wired second one.
-            """
-            callback = kwargs.get("progress_callback")
-            recorded.setdefault("calls", []).append(callback)
-            if callback:
-                callback()
-            return 0.05
-
     component = LoadMLComponent.__new__(LoadMLComponent)
     component.base = FakeBase()
     component.log = component.base.log
+    component.api_stop = False
     component.last_success_timestamp = None
-    component.predictor = FakePredictor()
+    component.predictor = predictor
     component.data_lock = asyncio.Lock()
     component.load_data = {1: 0.1}
     component.load_data_age_days = 10
@@ -3443,10 +3422,42 @@ def train_curriculum(self, *args, **kwargs):
     component.model_status = "not_initialized"
     component.last_train_time = None
     component.initial_training_done = False
+    return component
+
+
+def _test_component_marks_itself_alive_during_training():
+    """_do_training must hand the trainer the component's own timestamp updater.
+
+    Driven through the real _do_training rather than by calling train_curriculum directly - a test
+    that passes the callback itself proves only that the test passes it, and would keep passing if
+    load_ml_component stopped wiring it up. That regression is invisible until a training run
+    silently crosses the liveness window again hours later.
+    """
+    import asyncio
+
+    now_utc = datetime.now(timezone.utc)
+    recorded = {}
+
+    class FakePredictor:
+        """Captures the kwargs the component passes, and pulses the callback like a real run."""
+
+        def train_curriculum(self, *args, **kwargs):
+            """Record every call's progress callback and fire it once, as an epoch would.
+
+            Every call, not just the last: with is_initial the component runs the initial curriculum
+            and then a fine-tune pass, so keeping only the most recent kwargs would let an unwired
+            first call hide behind a correctly wired second one.
+            """
+            callback = kwargs.get("progress_callback")
+            recorded.setdefault("calls", []).append(callback)
+            if callback:
+                callback()
+            return 0.05
 
-    # Both paths, because _do_training picks a different train_curriculum call site for each - the
-    # initial curriculum or the fine-tune - and wiring only one of them up leaves the other able to
-    # blow past the liveness window unnoticed
+    component = _make_training_component(FakePredictor(), now_utc)
+
+    # Both paths, because _do_training sizes the initial curriculum and the fine-tune differently,
+    # and wiring only one of them up leaves the other able to blow past the liveness window unnoticed
     for is_initial in (False, True):
         recorded.clear()
         component.last_success_timestamp = None
@@ -3460,3 +3471,137 @@ def train_curriculum(self, *args, **kwargs):
         assert component.last_success_timestamp is not None, f"is_initial={is_initial}: firing the callback must move last_success_timestamp so components.is_alive() sees a recent success"
 
     return True
+
+
+def _test_component_training_does_not_block_event_loop():
+    """_do_training must leave the component's event loop free while training runs (#5075).
+
+    train_curriculum() is synchronous and minutes long, so calling it straight from the coroutine
+    pins the loop for the whole run: api_stop goes unnoticed and the run_timeout watchdog can never
+    fire. The handshake here is deterministic rather than timed - the fake trainer blocks until a
+    task scheduled on the loop releases it - so a loop that cannot run at all fails on the timeout
+    instead of on a stopwatch.
+    """
+    import asyncio
+    import threading
+
+    now_utc = datetime.now(timezone.utc)
+    loop_made_progress = threading.Event()
+    observed = {}
+
+    class BlockingPredictor:
+        """Blocks inside training the way a real curriculum pass does."""
+
+        def train_curriculum(self, *args, **kwargs):
+            """Wait for the loop to prove it is still running, then finish the 'run'."""
+            observed["loop_ran_during_training"] = loop_made_progress.wait(timeout=10.0)
+            return 0.05
+
+    component = _make_training_component(BlockingPredictor(), now_utc)
+
+    async def scenario():
+        """Queue a loop task, train, and see whether the task ever got a chance to run."""
+
+        async def releaser():
+            """Release the trainer, which only happens if the loop is still being serviced."""
+            loop_made_progress.set()
+
+        task = asyncio.create_task(releaser())
+        await component._do_training(is_initial=False)
+        await task
+
+    asyncio.run(scenario())
+
+    assert observed.get("loop_ran_during_training") is True, "the event loop made no progress while training ran - training is still running on the loop, so a shutdown landing mid-training cannot be answered"
+
+    return True
+
+
+def _test_component_training_abandons_run_on_stop():
+    """_do_training must let a stopping component abandon an in-flight training run (#5075).
+
+    Freeing the loop is not enough on its own: the component's thread only exits once run() returns,
+    so a curriculum that cannot be interrupted still holds the shutdown for as long as it takes to
+    finish. The component has to hand the trainer a stop hook that tracks api_stop, and must read the
+    resulting None as an abandoned run rather than a failed one - a partial model must not be saved
+    or published, and must not stamp last_train_time and so suppress the next real run.
+    """
+    import asyncio
+
+    now_utc = datetime.now(timezone.utc)
+    observed = {}
+
+    class StoppablePredictor:
+        """Polls the stop hook the way a real pass does, and abandons the run once it trips."""
+
+        def train_curriculum(self, *args, **kwargs):
+            """Record the hook's answer either side of the component being asked to stop."""
+            stop_callback = kwargs.get("stop_callback")
+            observed["hook"] = stop_callback
+            if stop_callback is None:
+                return 0.05
+            observed["before_stop"] = stop_callback()
+            component.api_stop = True
+            observed["after_stop"] = stop_callback()
+            return None if observed["after_stop"] else 0.05
+
+    component = _make_training_component(StoppablePredictor(), now_utc)
+    asyncio.run(component._do_training(is_initial=False))
+
+    assert observed.get("hook") is not None, "_do_training must pass a stop_callback to train_curriculum, or an in-flight run cannot be abandoned on shutdown"
+    assert observed.get("before_stop") is False, "the stop hook must report False while the component is still running"
+    assert observed.get("after_stop") is True, "the stop hook must report True once api_stop is set, so the run abandons itself"
+    assert component.model_valid is False, "an abandoned run must not be published as a valid model"
+    assert component.initial_training_done is False, "an abandoned run must not count as a completed training"
+    assert component.last_train_time is None, "an abandoned run must not stamp last_train_time, or the retrain interval skips the next real run"
+
+    return True
+
+
+def _test_training_stops_on_stop_callback():
+    """train() and train_curriculum() must abandon a run when the stop hook trips.
+
+    Deliberately a separate hook from progress_callback, which is barred from aborting a run: this
+    one is the caller explicitly asking training to give up. A hook that raises still must not.
+    """
+    now_utc = datetime.now(timezone.utc)
+    np.random.seed(42)
+    load_data = _create_synthetic_load_data(n_days=7, now_utc=now_utc)
+
+    # train() checks at the top of every epoch, so a hook that trips part-way ends the run there
+    epochs_completed = []
+    checks = {"count": 0}
+
+    def stop_on_third_check():
+        """Run two full epochs, then ask training to stop."""
+        checks["count"] += 1
+        return checks["count"] >= 3
+
+    predictor = LoadPredictor(learning_rate=0.01)
+    val_mae = predictor.train(load_data, now_utc, epochs=10, patience=10, progress_callback=lambda: epochs_completed.append(1), stop_callback=stop_on_third_check)
+    assert val_mae is None, "an abandoned run must report None so the caller neither saves nor publishes a partial model"
+    assert len(epochs_completed) == 2, f"training should stop at the epoch the hook first trips, so 2 epochs complete, got {len(epochs_completed)}"
+
+    # A hook that raises must not abort the run, same rule the progress hook follows
+    def bad_stop():
+        """Blow up the way a badly behaved caller's hook might."""
+        raise RuntimeError("stop hook blew up")
+
+    predictor_bad = LoadPredictor(learning_rate=0.01)
+    val_mae_bad = predictor_bad.train(load_data, now_utc, epochs=3, patience=3, stop_callback=bad_stop)
+    assert val_mae_bad is not None, "training must survive a stop hook that raises"
+
+    # Curriculum abandons the whole run rather than carrying on into the next pass
+    load_data_28 = _create_synthetic_load_data(n_days=28, now_utc=now_utc)
+    passes_started = []
+    predictor_c = LoadPredictor(learning_rate=0.01)
+    predictor_c.train = lambda *args, **kwargs: passes_started.append(1)
+    curriculum_mae = predictor_c.train_curriculum(load_data_28, now_utc, epochs=3, patience=3, curriculum_window_days=7, curriculum_step_days=7, stop_callback=lambda: True)
+    assert curriculum_mae is None, "a curriculum abandoned before its first pass must return None"
+    assert passes_started == [], "no pass should be started once the stop hook has tripped"
+
+    # Omitting the hook must stay valid - every existing caller relies on that
+    predictor_none = LoadPredictor(learning_rate=0.01)
+    assert predictor_none.train(load_data, now_utc, epochs=2, patience=3) is not None, "training without a stop hook must be unaffected"
+
+    return True

From be298f428a2ccb13ad5eca082ce1ad38fb84bfa3 Mon Sep 17 00:00:00 2001
From: CI <ci@example.com>
Date: Thu, 24 Sep 2026 09:04:03 +0100
Subject: [PATCH 2/5] Stop a cancelled ML training run orphaning its worker
 thread

Addresses the review on #5112. Now that training runs off the loop, the
run_timeout watchdog can actually fire, and its task.cancel() reaches only
the coroutine - not the thread already inside train_curriculum.

- _do_training raises the component's own cancel signal on CancelledError
  and the worker thread owns an in-flight flag, so a cancelled run is
  abandoned rather than left racing a second curriculum on the same
  predictor's weights, Adam state and model file.
- training_stop_requested also answers to fatal_error, which exits
  ComponentBase.start() without ever calling stop().
- The abandon-vs-failure split is decided by whether the stop hook actually
  answered True during the run, not by re-reading api_stop afterwards, so a
  genuine all-passes-failed run is still warned about and counted.
- run() publishes the model status on the stop path (training may have
  completed before the stop landed) and no longer waits out a prediction
  cycle it is about to abandon.
- train() restores the normalisation statistics alongside the weights on an
  abort, and skips the minutes-long AR rollout diagnostic when a stop lands
  after the last epoch check.
- train_curriculum rolls back the published training stamps when a run is
  abandoned mid-curriculum, with a new model_trained flag so a never-trained
  model cannot read as active while a legacy timestamp-less model still can.
- A stop hook that raises is reported once at Error level instead of being
  warned about once per epoch.

Five new tests in test_load_ml.py, all verified red without these changes,
plus a stronger handshake in training_off_event_loop that cannot pass on a
single incidental loop yield.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
---
 apps/predbat/load_ml_component.py  | 123 ++++++++++----
 apps/predbat/load_predictor.py     |  89 ++++++++--
 apps/predbat/tests/test_load_ml.py | 262 ++++++++++++++++++++++++++++-
 3 files changed, 424 insertions(+), 50 deletions(-)

diff --git a/apps/predbat/load_ml_component.py b/apps/predbat/load_ml_component.py
index 0756f4653..4e84685b7 100644
--- a/apps/predbat/load_ml_component.py
+++ b/apps/predbat/load_ml_component.py
@@ -121,6 +121,13 @@ def initialize(self, load_ml_enable, load_ml_source=True, load_ml_max_days_histo
         self.initial_training_done = False
         self.database_history_loaded = False
 
+        # In-flight training state. training_running is owned by the worker thread (it clears it
+        # when the synchronous curriculum truly returns, not when the awaiting coroutine goes away),
+        # and training_cancelled is the component's own cancel signal for a run whose coroutine has
+        # been cancelled out from under it - see _do_training.
+        self.training_running = False
+        self.training_cancelled = False
+
         # Predictions cache
         self.current_predictions = {}
 
@@ -768,21 +775,26 @@ async def run(self, seconds, first):
             # via is_calculating() and logged by calculate_plan().
             if self.base.prediction_started:
                 self.log("ML Component: Waiting for current prediction cycle to complete before running ML work")
-                while self.base.prediction_started:
+                # api_stop is part of the condition because a plan cycle is itself minutes long: a
+                # stop landing while one runs would otherwise be held here until it finished, which
+                # is the same delay class #5075 set out to remove.
+                while self.base.prediction_started and not self.api_stop:
                     await asyncio.sleep(0.5)
             try:
                 self.load_ml_calculating = True
 
-                if should_train:
+                if should_train and not self.api_stop:
                     self.log("ML Component: Doing training...")
                     await self._do_training(is_initial)
 
-                    if self.api_stop:
-                        # Training abandoned for shutdown - a re-fetch and a prediction cycle here
-                        # would just add their own delay to the stop we are already answering
-                        self.log("ML Component: Stopping, skipping the post-training prediction cycle")
-                        return True
-
+                if self.api_stop:
+                    # A re-fetch and a prediction cycle here would just add their own delay to the
+                    # stop we are already answering. The model status below still runs: training may
+                    # have completed successfully before the stop landed, and skipping it entirely
+                    # would leave that fresh model unpublished.
+                    self.log("ML Component: Stopping, skipping the post-training prediction cycle")
+                    should_fetch = False
+                elif should_train:
                     # Training can run for many minutes, during which the data fetched above
                     # goes stale - both the lookback window feeding the prediction and the
                     # load_minutes_now baseline still refer to the pre-training time. Re-fetch
@@ -952,8 +964,14 @@ def array_to_dict(arr):
             self.data_ready = False
 
     def training_stop_requested(self):
-        """Return True when the component is stopping, so an in-flight training run abandons itself."""
-        return self.api_stop
+        """Return True when this training run has been cancelled, so the worker thread abandons it.
+
+        Three sources, because a training run outlives its coroutine: api_stop for an ordinary
+        shutdown, fatal_error because ComponentBase.start() exits on it without ever calling stop(),
+        and training_cancelled for a run() the run_timeout watchdog has cancelled - in that last case
+        api_stop is still False, so without this the worker thread would train on unwatched.
+        """
+        return self.api_stop or self.fatal_error or self.training_cancelled
 
     async def _do_training(self, is_initial):
         """
@@ -962,6 +980,15 @@ async def _do_training(self, is_initial):
         Args:
             is_initial: True for full training, False for fine-tuning
         """
+        if self.training_running:
+            # A previous run()'s worker thread is still inside train_curriculum. That happens when
+            # the run_timeout watchdog cancelled the coroutine awaiting it: the coroutine went away,
+            # the thread did not. Starting a second curriculum here would have two threads mutating
+            # the same LoadPredictor's weights, biases, Adam moments and normalisation statistics in
+            # place, and both racing to save() over the same model file.
+            self.log("Warn: ML Component: Previous training run is still finishing, skipping this training cycle")
+            return
+
         # Snapshot data under the lock so we can release it before the CPU-bound
         # training call.  A curriculum run is many passes of many epochs and routinely takes
         # minutes, so it must NOT run while holding data_lock, and must not run on the event
@@ -997,33 +1024,67 @@ async def _do_training(self, is_initial):
                 max_intermediate_passes = self.ml_curriculum_max_passes
         # Lock released
 
+        # Tracks whether the stop hook ever actually answered True during this run. train() and
+        # train_curriculum() both return None for "aborted" and for "every pass failed", so reading
+        # the hook again after the fact would file a genuine data failure as an abandonment whenever
+        # a stop happened to land in between - and then never count the error or warn about it.
+        stop_signalled = {"tripped": False}
+
+        def stop_hook():
+            """Answer the trainer's abort question, remembering a True so the caller can tell why."""
+            if self.training_stop_requested():
+                stop_signalled["tripped"] = True
+                return True
+            return False
+
+        def run_training():
+            """Run the curriculum on this worker thread, owning the in-flight flag for its lifetime.
+
+            The flag is raised and cleared here rather than in the awaiting coroutine because a
+            cancelled coroutine never reaches its own cleanup, while this thread always finishes.
+            """
+            self.training_running = True
+            try:
+                return self.predictor.train_curriculum(
+                    load_data_snap,
+                    now_utc_snap,
+                    pv_minutes=pv_data_snap,
+                    temp_minutes=temp_data_snap,
+                    import_rates=import_rates_snap,
+                    export_rates=export_rates_snap,
+                    epochs=epochs,
+                    time_decay_days=time_decay,
+                    validation_holdout_hours=holdout_hours,
+                    patience=patience,
+                    curriculum_window_days=window_days,
+                    curriculum_step_days=step_days,
+                    max_intermediate_passes=max_intermediate_passes,
+                    progress_callback=self.update_success_timestamp,
+                    stop_callback=stop_hook,
+                )
+            finally:
+                self.training_running = False
+
+        self.training_cancelled = False
         try:
             # Run the curriculum on a worker thread. It is synchronous, CPU-bound and minutes long,
             # so calling it directly would pin this component's event loop for its whole duration:
             # api_stop would go unnoticed, the run_timeout watchdog could never fire, and a shutdown
             # landing mid-training would sit waiting on this component's thread (#5075).
-            val_mae = await asyncio.to_thread(
-                self.predictor.train_curriculum,
-                load_data_snap,
-                now_utc_snap,
-                pv_minutes=pv_data_snap,
-                temp_minutes=temp_data_snap,
-                import_rates=import_rates_snap,
-                export_rates=export_rates_snap,
-                epochs=epochs,
-                time_decay_days=time_decay,
-                validation_holdout_hours=holdout_hours,
-                patience=patience,
-                curriculum_window_days=window_days,
-                curriculum_step_days=step_days,
-                max_intermediate_passes=max_intermediate_passes,
-                progress_callback=self.update_success_timestamp,
-                stop_callback=self.training_stop_requested,
-            )
-
-            if val_mae is None and self.training_stop_requested():
+            try:
+                val_mae = await asyncio.to_thread(run_training)
+            except asyncio.CancelledError:
+                # Now that the loop is free the run_timeout watchdog can fire, and cancelling this
+                # coroutine does nothing to the worker thread already inside train_curriculum.
+                # Trip our own stop signal so the thread abandons itself at its next epoch check
+                # instead of training on unwatched, then let the cancellation propagate.
+                self.training_cancelled = True
+                self.log("Warn: ML Component: Training run was cancelled (run_timeout is {}s) - asking the training thread to abandon the run".format(self.run_timeout))
+                raise
+
+            if val_mae is None and stop_signalled["tripped"]:
                 # Abandoned on purpose, not a failure - and nothing to save, the model is partial
-                self.log("ML Component: Training abandoned because the component is stopping")
+                self.log("ML Component: Training abandoned because the run was cancelled or the component is stopping")
                 return
 
             if val_mae is not None:
diff --git a/apps/predbat/load_predictor.py b/apps/predbat/load_predictor.py
index a89196337..5f0c9343a 100644
--- a/apps/predbat/load_predictor.py
+++ b/apps/predbat/load_predictor.py
@@ -221,6 +221,11 @@ def __init__(self, log_func=None, learning_rate=0.001, max_load_kw=23.0, weight_
         self.pv_std = None
 
         # Training metadata
+        self.stop_callback_failures = 0  # Consecutive raises from a caller's stop hook, see _should_stop
+        # True once a train() has run to completion, or once a saved model has been loaded (save()
+        # only ever follows a completed training). Distinct from model_initialized, which is already
+        # True while the weights are still the random initialisation - see is_valid().
+        self.model_trained = False
         self.training_timestamp = None
         self.validation_mae = None
         self.validation_bias = None  # Signed metric: mean(predicted - actual); + = over-predicting, - = under-predicting
@@ -1171,13 +1176,24 @@ def _clip_predictions(self, predictions, lookback_buffer=None):
         return predictions
 
     def _should_stop(self, stop_callback):
-        """Return True when the caller has asked training to abort, treating a raising hook as no."""
+        """Return True when the caller has asked training to abort, treating a raising hook as no.
+
+        A raising hook is not allowed to kill a run - same rule as progress_callback - but unlike
+        that one, a permanently broken hook here silently costs the caller its whole abort mechanism
+        and turns a shutdown back into a full-length curriculum. The hook is polled once per epoch,
+        so the failure is counted and reported once at Error level rather than warned about hundreds
+        of times, which would bury it.
+        """
         if not stop_callback:
             return False
         try:
-            return bool(stop_callback())
+            result = bool(stop_callback())
+            self.stop_callback_failures = 0
+            return result
         except Exception as e:
-            self.log("Warn: ML Predictor: stop callback raised {}, continuing training".format(e))
+            self.stop_callback_failures += 1
+            if self.stop_callback_failures == 1:
+                self.log("Error: ML Predictor: stop callback raised {}, continuing training - this run can no longer be aborted".format(e))
             return False
 
     def train(
@@ -1258,6 +1274,15 @@ def train(
             self.log("Warn: ML Predictor: No validation data available")
             return None
 
+        # Snapshot the normalisation statistics before they are refitted below. An aborted run
+        # restores its starting weights, but weights and normalisation are one model: leaving the
+        # newly fitted statistics in place alongside restored weights would have every subsequent
+        # predict() scale its inputs by statistics those weights were never trained against.
+        entry_feature_mean = self.feature_mean.copy() if self.feature_mean is not None else None
+        entry_feature_std = self.feature_std.copy() if self.feature_std is not None else None
+        entry_target_mean = self.target_mean
+        entry_target_std = self.target_std
+
         # Normalize features and targets
         # On initial train: fit normalization from scratch
         # On fine-tune: apply EMA update to track distribution drift gradually
@@ -1443,11 +1468,18 @@ def train(
             )
 
         if aborted:
-            # Weights are restored above so the in-memory model is left consistent, but the run did
-            # not finish: report failure so the caller neither publishes nor saves a partial model,
-            # and skip the AR rollout diagnostic, which is itself minutes of the work we just abandoned.
+            # Weights are restored above; put the normalisation statistics back with them so the
+            # in-memory model is left wholly as it was found, rather than as restored weights paired
+            # with statistics refitted for the training that was abandoned. Report failure so the
+            # caller neither publishes nor saves a partial model, and skip the AR rollout diagnostic,
+            # which is itself minutes of the work we just abandoned.
+            self.feature_mean = entry_feature_mean
+            self.feature_std = entry_feature_std
+            self.target_mean = entry_target_mean
+            self.target_std = entry_target_std
             return None
 
+        self.model_trained = True
         self.training_timestamp = datetime.now(timezone.utc)
         self.validation_mae = best_val_loss
         self.validation_bias = float(best_val_bias)
@@ -1458,6 +1490,14 @@ def train(
 
         # Autoregressive diagnostic: run a full AR rollout over the holdout period
         # to expose compounding error (teacher-forced val_mae won't show this)
+        if self._should_stop(stop_callback):
+            # A stop that lands after the last epoch check still has to get past the rollout, which
+            # is itself minutes of work and purely diagnostic. The training itself did complete, so
+            # the result is returned and the caller may publish and save it - only the diagnostic is
+            # given up, leaving rollout_mae/pattern_mae as whatever the last completed run set.
+            self.log("ML Predictor: Skipping AR rollout diagnostic - stop requested")
+            return best_val_loss
+
         ar_mae, ar_bias, pattern_mae = self._ar_rollout_diagnostic(
             load_minutes,
             now_utc,
@@ -1575,6 +1615,20 @@ def train_curriculum(
             self.log("ML Predictor: Curriculum training not started - stop requested")
             return None
 
+        # Snapshot the published training stamps. train() stamps these at the end of EVERY completed
+        # pass, including the intermediate curriculum windows, so an abandoned curriculum would
+        # otherwise leave the last completed intermediate pass's timestamp and validation_mae behind
+        # - and is_valid() judges purely on those, so a model trained on the first 7 days of a 28-day
+        # curriculum would report "active" and its fresh timestamp would suppress the staleness
+        # retrain that is the only route back to a complete model.
+        entry_stamps = (self.model_trained, self.training_timestamp, self.validation_mae, self.validation_bias, self.epochs_trained)
+
+        def abandon_curriculum(message):
+            """Log an abandoned curriculum, roll the published stamps back, and report no result."""
+            self.log(message)
+            self.model_trained, self.training_timestamp, self.validation_mae, self.validation_bias, self.epochs_trained = entry_stamps
+            return None
+
         # Build list of positive minute keys to find total history span
         hist_minutes = [k for k in load_minutes if isinstance(k, int) and k > 0]
         if not hist_minutes:
@@ -1628,8 +1682,7 @@ def train_curriculum(
         val_mae = None
         for pass_idx, window in enumerate(window_sizes):
             if self._should_stop(stop_callback):
-                self.log("ML Predictor: Curriculum training aborted before pass {}/{} - stop requested".format(pass_idx + 1, total_passes))
-                return None
+                return abandon_curriculum("ML Predictor: Curriculum training aborted before pass {}/{} - stop requested".format(pass_idx + 1, total_passes))
 
             # Slice data to the oldest 'window' minutes.
             # start_minute is the more-recent edge; after re-indexing it becomes key 0.
@@ -1665,8 +1718,7 @@ def train_curriculum(
             )
 
             if pass_mae is None and self._should_stop(stop_callback):
-                self.log("ML Predictor: Curriculum training aborted during pass {}/{} - stop requested".format(pass_idx + 1, total_passes))
-                return None
+                return abandon_curriculum("ML Predictor: Curriculum training aborted during pass {}/{} - stop requested".format(pass_idx + 1, total_passes))
 
             if pass_mae is None:
                 self.log("Warn: ML Predictor: Curriculum pass {}/{} failed (insufficient data or training error) - skipping".format(pass_idx + 1, total_passes))
@@ -1676,8 +1728,7 @@ def train_curriculum(
 
         # Final pass: full dataset, standard holdout window
         if self._should_stop(stop_callback):
-            self.log("ML Predictor: Curriculum training aborted before final pass {}/{} - stop requested".format(total_passes, total_passes))
-            return None
+            return abandon_curriculum("ML Predictor: Curriculum training aborted before final pass {}/{} - stop requested".format(total_passes, total_passes))
 
         self.log("ML Predictor: Curriculum final pass {}/{}: full dataset ({:.1f} days)".format(total_passes, total_passes, max_minute / day_minutes))
         final_mae = self.train(
@@ -1699,8 +1750,7 @@ def train_curriculum(
         )
 
         if final_mae is None and self._should_stop(stop_callback):
-            self.log("ML Predictor: Curriculum training aborted during final pass {}/{} - stop requested".format(total_passes, total_passes))
-            return None
+            return abandon_curriculum("ML Predictor: Curriculum training aborted during final pass {}/{} - stop requested".format(total_passes, total_passes))
 
         if final_mae is not None:
             val_mae = final_mae
@@ -2076,6 +2126,9 @@ def load(self, filepath):
             self.dropout_rate = metadata.get("dropout_rate", 0.1)
 
             self.model_initialized = True
+            # A model only ever reaches disk after a completed training, so a loaded one counts as
+            # trained even when it predates training_timestamp being saved
+            self.model_trained = True
 
             self.log(
                 "ML Predictor: Model loaded from {} (trained {}, val_mae={:.4f}, val_bias={:+.4f})".format(
@@ -2113,6 +2166,14 @@ def is_valid(self, validation_threshold=2.0, max_age_hours=48):
         if self.weights is None:
             return False, "no_weights"
 
+        if not self.model_trained:
+            # model_initialized only means _initialize_weights() has run, which it does before the
+            # first epoch, so weights alone can still be the random He initialisation. Without this
+            # an initial run abandoned mid-curriculum - whose stamps train_curriculum rolls back -
+            # would report "active" on a model that has never been trained. Tested separately from
+            # training_timestamp, which a model saved before timestamps existed legitimately lacks.
+            return False, "not_trained"
+
         if self.validation_mae is not None and self.validation_mae > validation_threshold:
             return False, "validation_threshold"
 
diff --git a/apps/predbat/tests/test_load_ml.py b/apps/predbat/tests/test_load_ml.py
index 26944e2d8..23740f945 100644
--- a/apps/predbat/tests/test_load_ml.py
+++ b/apps/predbat/tests/test_load_ml.py
@@ -59,6 +59,11 @@ def test_load_ml(my_predbat=None):
         ("training_stop_callback", _test_training_stops_on_stop_callback, "Training abandons a run when the stop hook trips"),
         ("training_off_event_loop", _test_component_training_does_not_block_event_loop, "Component training leaves its event loop free to run"),
         ("training_abandoned_on_stop", _test_component_training_abandons_run_on_stop, "Component abandons an in-flight training run when stopping"),
+        ("training_abort_restores_norm", _test_training_abort_restores_normalisation, "An aborted train() restores normalisation statistics with the weights"),
+        ("curriculum_mid_abort_rollback", _test_curriculum_mid_abort_rolls_back_stamps, "An abandoned curriculum rolls back the completed passes' published stamps"),
+        ("untrained_model_not_valid", _test_untrained_model_is_not_valid, "A model that never completed a training does not read as valid"),
+        ("training_watchdog_cancel", _test_component_training_survives_watchdog_cancel, "A cancelled run() stops the training thread instead of orphaning it"),
+        ("training_failure_not_abandon", _test_component_training_failure_is_not_reported_as_abandoned, "An all-passes-failed run is reported as a failure, not an abandonment"),
         ("prediction", _test_prediction, "End-to-end prediction"),
         ("prediction_with_pv", _test_prediction_with_pv, "Prediction with PV forecast data"),
         ("prediction_with_temp", _test_prediction_with_temp, "Prediction with temperature forecast data"),
@@ -3389,6 +3394,7 @@ class FakeBase:
         def __init__(self):
             """Expose only what _do_training touches."""
             self.now_utc = now_utc
+            self.fatal_error = False
 
         def log(self, msg, **kwargs):
             """Swallow component logging."""
@@ -3422,6 +3428,9 @@ def log(self, msg, **kwargs):
     component.model_status = "not_initialized"
     component.last_train_time = None
     component.initial_training_done = False
+    component.run_timeout = 2 * 60 * 60
+    component.training_running = False
+    component.training_cancelled = False
     return component
 
 
@@ -3486,15 +3495,25 @@ def _test_component_training_does_not_block_event_loop():
     import threading
 
     now_utc = datetime.now(timezone.utc)
-    loop_made_progress = threading.Event()
+    # Two round trips, each gated on the trainer having reached that point. A single one-shot event
+    # released unconditionally would pass spuriously: any earlier await in _do_training that happens
+    # to yield once sets it before the trainer is entered, after which a regression back to a
+    # synchronous loop-pinning call still finds it set. Each of these is only set by a loop task
+    # that ran *after* the trainer said it was inside the call.
+    trainer_reached = [threading.Event(), threading.Event()]
+    loop_released = [threading.Event(), threading.Event()]
     observed = {}
 
     class BlockingPredictor:
         """Blocks inside training the way a real curriculum pass does."""
 
         def train_curriculum(self, *args, **kwargs):
-            """Wait for the loop to prove it is still running, then finish the 'run'."""
-            observed["loop_ran_during_training"] = loop_made_progress.wait(timeout=10.0)
+            """Hand control to the loop twice from inside the 'run', then finish it."""
+            released = []
+            for round_index in range(2):
+                trainer_reached[round_index].set()
+                released.append(loop_released[round_index].wait(timeout=10.0))
+            observed["loop_ran_during_training"] = all(released)
             return 0.05
 
     component = _make_training_component(BlockingPredictor(), now_utc)
@@ -3503,8 +3522,17 @@ async def scenario():
         """Queue a loop task, train, and see whether the task ever got a chance to run."""
 
         async def releaser():
-            """Release the trainer, which only happens if the loop is still being serviced."""
-            loop_made_progress.set()
+            """Answer each of the trainer's handshakes, giving up rather than spinning forever."""
+            for round_index in range(2):
+                for _ in range(200000):
+                    if trainer_reached[round_index].is_set():
+                        break
+                    await asyncio.sleep(0)
+                else:
+                    observed["gave_up_on_round"] = round_index
+                    return
+                loop_released[round_index].set()
+            observed["loop_answered_both_rounds"] = True
 
         task = asyncio.create_task(releaser())
         await component._do_training(is_initial=False)
@@ -3513,6 +3541,7 @@ async def releaser():
     asyncio.run(scenario())
 
     assert observed.get("loop_ran_during_training") is True, "the event loop made no progress while training ran - training is still running on the loop, so a shutdown landing mid-training cannot be answered"
+    assert observed.get("loop_answered_both_rounds") is True, "a loop task did not run between the trainer's two handshakes, so the loop is not being serviced for the duration of the training call"
 
     return True
 
@@ -3558,6 +3587,229 @@ def train_curriculum(self, *args, **kwargs):
     return True
 
 
+def _test_training_abort_restores_normalisation():
+    """An aborted train() must put back the normalisation statistics along with the weights.
+
+    Weights and normalisation are one model. train() refits the statistics before the first epoch
+    check, so restoring only best_weights would leave restored weights paired with statistics fitted
+    for the run that was abandoned, and every subsequent predict() would scale its inputs by numbers
+    those weights were never trained against. Only reachable while the component survives the abort
+    - a cancelled run(), not a process shutdown - which is exactly the run_timeout watchdog path.
+    """
+    now_utc = datetime.now(timezone.utc)
+    np.random.seed(11)
+    load_data = _create_synthetic_load_data(n_days=7, now_utc=now_utc)
+
+    predictor = LoadPredictor(learning_rate=0.01)
+    assert predictor.train(load_data, now_utc, epochs=2, patience=3) is not None, "the baseline run must succeed so there is a fitted model to abort against"
+
+    before = {
+        "feature_mean": predictor.feature_mean.copy(),
+        "feature_std": predictor.feature_std.copy(),
+        "target_mean": predictor.target_mean,
+        "target_std": predictor.target_std,
+        "weights": [w.copy() for w in predictor.weights],
+    }
+
+    # Fine-tune on plainly different data so the EMA refit moves the statistics by an amount no
+    # tolerance could absorb, then abort at the very first epoch check - after the refit has
+    # already happened
+    other_data = {minute: value * 5.0 for minute, value in load_data.items()}
+    aborted_mae = predictor.train(other_data, now_utc, epochs=5, patience=3, stop_callback=lambda: True)
+
+    assert aborted_mae is None, "an aborted fine-tune must report None"
+    assert np.allclose(predictor.feature_mean, before["feature_mean"]), "an aborted run must restore feature_mean, or predictions are normalised by statistics the restored weights never saw"
+    assert np.allclose(predictor.feature_std, before["feature_std"]), "an aborted run must restore feature_std"
+    assert predictor.target_mean == before["target_mean"], "an aborted run must restore target_mean"
+    assert predictor.target_std == before["target_std"], "an aborted run must restore target_std"
+    for n, (restored, original) in enumerate(zip(predictor.weights, before["weights"])):
+        assert np.allclose(restored, original), f"an aborted run must restore layer {n} weights unchanged"
+
+    return True
+
+
+def _test_curriculum_mid_abort_rolls_back_stamps():
+    """A curriculum abandoned mid-run must not leave a completed pass's stamps published.
+
+    train() stamps training_timestamp/validation_mae/epochs_trained at the end of every pass,
+    including the intermediate curriculum windows. is_valid() judges purely on those two, so without
+    a rollback a model trained on only the first window of the curriculum reports "active" and its
+    fresh timestamp suppresses the staleness retrain that is the only route back to a full model.
+    """
+    now_utc = datetime.now(timezone.utc)
+    np.random.seed(13)
+    load_data = _create_synthetic_load_data(n_days=28, now_utc=now_utc)
+
+    stale_timestamp = now_utc - timedelta(hours=100)
+    predictor = LoadPredictor(learning_rate=0.01)
+    predictor.model_initialized = True
+    predictor.model_trained = True
+    predictor.weights = [np.zeros((1, 1), dtype=np.float32)]
+    predictor.training_timestamp = stale_timestamp
+    predictor.validation_mae = 0.4
+    predictor.validation_bias = 0.05
+    predictor.epochs_trained = 42
+
+    assert predictor.is_valid(validation_threshold=2.0, max_age_hours=48) == (False, "stale"), "the starting model must be due a retrain, so the abandonment cannot be mistaken for it"
+
+    passes = {"count": 0}
+
+    def fake_train(*args, **kwargs):
+        """Stand in for a completed pass, stamping the model the way the real train() does."""
+        passes["count"] += 1
+        predictor.model_trained = True
+        predictor.training_timestamp = now_utc
+        predictor.validation_mae = 0.3
+        predictor.validation_bias = 0.01
+        predictor.epochs_trained += 3
+        return 0.3
+
+    predictor.train = fake_train
+    # Trips only once a pass has completed, so the run is abandoned between passes rather than
+    # before the first one - the path the existing stop-callback test never reaches
+    curriculum_mae = predictor.train_curriculum(load_data, now_utc, epochs=3, patience=3, curriculum_window_days=7, curriculum_step_days=7, stop_callback=lambda: passes["count"] >= 1)
+
+    assert passes["count"] == 1, f"exactly one pass should complete before the hook trips, got {passes['count']}"
+    assert curriculum_mae is None, "an abandoned curriculum must not return the completed intermediate pass's val_mae as a finished result"
+    assert predictor.training_timestamp == stale_timestamp, "an abandoned curriculum must roll back training_timestamp, or a partial model's fresh stamp suppresses the staleness retrain"
+    assert predictor.validation_mae == 0.4, "an abandoned curriculum must roll back validation_mae"
+    assert predictor.validation_bias == 0.05, "an abandoned curriculum must roll back validation_bias"
+    assert predictor.epochs_trained == 42, "an abandoned curriculum must roll back epochs_trained"
+    assert predictor.is_valid(validation_threshold=2.0, max_age_hours=48) == (False, "stale"), "the model must still read as due a retrain after an abandoned curriculum"
+
+    return True
+
+
+def _test_untrained_model_is_not_valid():
+    """A model whose weights exist but which has never completed a training must not read as active.
+
+    _initialize_weights() sets model_initialized before the first epoch, so an initial run abandoned
+    part-way leaves random He-initialised weights behind. model_trained is the separate signal that
+    a training actually landed - kept apart from training_timestamp, which a model saved before
+    timestamps were stored legitimately lacks while still being a properly trained model.
+    """
+    predictor = LoadPredictor(learning_rate=0.01)
+    predictor.model_initialized = True
+    predictor.weights = [np.zeros((1, 1), dtype=np.float32)]
+    predictor.training_timestamp = None
+    predictor.validation_mae = None
+
+    assert predictor.is_valid(validation_threshold=2.0, max_age_hours=48) == (False, "not_trained"), "a model that has never completed a training must not be published as valid"
+
+    # A legacy saved model has no timestamp but has been trained, and must keep serving forecasts
+    predictor.model_trained = True
+    assert predictor.is_valid(validation_threshold=2.0, max_age_hours=48) == (True, None), "a trained model without an embedded timestamp must still be valid"
+
+    return True
+
+
+def _test_component_training_survives_watchdog_cancel():
+    """A cancelled run() must stop the training thread, not orphan it (#5075 follow-up).
+
+    Now that training is off the loop, ComponentBase.start()'s run_timeout watchdog can actually
+    fire, and its task.cancel() reaches the coroutine awaiting to_thread - not the worker thread
+    already inside train_curriculum. api_stop is still False on that path, so without the
+    component's own cancel signal the thread would train on unwatched, and the next 60s tick would
+    start a second curriculum mutating the same predictor's weights and Adam state in place.
+    """
+    import asyncio
+    import threading
+    import time
+
+    now_utc = datetime.now(timezone.utc)
+    entered = threading.Event()
+    allow_exit = threading.Event()
+    observed = {"calls": 0}
+
+    class CancelWatchPredictor:
+        """Polls the stop hook until it trips, then holds the thread open so the guard can be seen."""
+
+        def train_curriculum(self, *args, **kwargs):
+            """Record that the hook tripped, and stay 'in training' until released."""
+            stop_callback = kwargs.get("stop_callback")
+            observed["calls"] += 1
+            entered.set()
+            for _ in range(1000):
+                if stop_callback():
+                    observed["hook_tripped"] = True
+                    break
+                time.sleep(0.01)
+            allow_exit.wait(timeout=10.0)
+            return None
+
+    component = _make_training_component(CancelWatchPredictor(), now_utc)
+
+    async def scenario():
+        """Cancel the training coroutine mid-run, then try to start another cycle."""
+        task = asyncio.ensure_future(component._do_training(is_initial=False))
+        for _ in range(200000):
+            if entered.is_set():
+                break
+            await asyncio.sleep(0)
+        assert entered.is_set(), "the trainer was never entered, so there is nothing to cancel"
+
+        task.cancel()
+        try:
+            await task
+        except asyncio.CancelledError:
+            observed["cancelled"] = True
+
+        # The worker thread is still inside train_curriculum here. The next run() tick must not
+        # start a second one on the same predictor.
+        await component._do_training(is_initial=False)
+        allow_exit.set()
+
+        for _ in range(1000):
+            if not component.training_running:
+                break
+            await asyncio.sleep(0.01)
+
+    asyncio.run(scenario())
+
+    assert observed.get("cancelled") is True, "cancelling the run() task must propagate out of _do_training rather than being swallowed"
+    assert component.training_cancelled is True, "a cancelled run must raise the component's own cancel signal, or the worker thread never learns to abandon"
+    assert observed.get("hook_tripped") is True, "the stop hook must report True after the coroutine is cancelled, so the orphaned training thread abandons its run"
+    assert observed["calls"] == 1, f"a second training must not start while the previous worker thread is still running, got {observed['calls']} calls"
+    assert component.training_running is False, "the worker thread must clear training_running when it returns, or training never runs again"
+    assert component.last_train_time is None, "a cancelled run must not stamp last_train_time"
+
+    return True
+
+
+def _test_component_training_failure_is_not_reported_as_abandoned():
+    """A genuinely failed training must be reported as a failure even if a stop lands afterwards.
+
+    train_curriculum() returns None both for "aborted" and for "every pass failed", so re-reading
+    api_stop after the fact files a real data problem as an abandonment - and then never warns about
+    it or counts the error. The component has to remember whether its stop hook actually answered
+    True during the run.
+    """
+    import asyncio
+
+    now_utc = datetime.now(timezone.utc)
+    logged = []
+
+    class FailingPredictor:
+        """Fails every pass without the stop hook ever tripping, then a stop lands at the end."""
+
+        def train_curriculum(self, *args, **kwargs):
+            """Return the all-passes-failed None, with api_stop set just as the run finishes."""
+            component.api_stop = True
+            return None
+
+    component = _make_training_component(FailingPredictor(), now_utc)
+    component.log = lambda msg, **kwargs: logged.append(msg)
+
+    asyncio.run(component._do_training(is_initial=False))
+
+    joined = " | ".join(logged)
+    assert "Training failed" in joined, f"an all-passes-failed run must still be reported as a failure, logged: {joined}"
+    assert "Training abandoned" not in joined, f"a run whose stop hook never tripped must not be filed as an abandonment, logged: {joined}"
+    assert component.last_train_time is None, "a failed run must not stamp last_train_time"
+
+    return True
+
+
 def _test_training_stops_on_stop_callback():
     """train() and train_curriculum() must abandon a run when the stop hook trips.
 

From c6af1e17ce228d303e9bf12aca283c607eac9407 Mon Sep 17 00:00:00 2001
From: CI <ci@example.com>
Date: Thu, 24 Sep 2026 09:09:19 +0100
Subject: [PATCH 3/5] Report is_calculating() from the training thread, not the
 cancelled coroutine

run()'s finally clears load_ml_calculating while a watchdog-cancelled run's
worker thread is still inside train_curriculum, so the flag alone said no
NumPy work was in flight during exactly that work. is_calculating() now also
answers to training_running, which the worker thread owns for the span of the
synchronous call.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
---
 apps/predbat/load_ml_component.py  | 9 +++++++--
 apps/predbat/tests/test_load_ml.py | 9 +++++++--
 2 files changed, 14 insertions(+), 4 deletions(-)

diff --git a/apps/predbat/load_ml_component.py b/apps/predbat/load_ml_component.py
index 4e84685b7..45d53d15e 100644
--- a/apps/predbat/load_ml_component.py
+++ b/apps/predbat/load_ml_component.py
@@ -177,8 +177,13 @@ def _init_predictor(self):
                 self.predictor = LoadPredictor(log_func=self.log, learning_rate=self.ml_learning_rate, max_load_kw=self.ml_max_load_kw, weight_decay=self.ml_weight_decay, dropout_rate=self.ml_dropout_rate)
 
     def is_calculating(self):
-        """Return whether the component is currently calculating predictions."""
-        return self.load_ml_calculating
+        """Return whether the component is currently doing NumPy-heavy work.
+
+        training_running is part of the answer because a training run outlives the coroutine that
+        started it: when the run_timeout watchdog cancels run(), its finally clears
+        load_ml_calculating while the worker thread is still inside train_curriculum.
+        """
+        return self.load_ml_calculating or self.training_running
 
     def get_from_incrementing(self, data, index, step, backwards=True):
         """
diff --git a/apps/predbat/tests/test_load_ml.py b/apps/predbat/tests/test_load_ml.py
index 23740f945..cc7fcb833 100644
--- a/apps/predbat/tests/test_load_ml.py
+++ b/apps/predbat/tests/test_load_ml.py
@@ -3429,6 +3429,7 @@ def log(self, msg, **kwargs):
     component.last_train_time = None
     component.initial_training_done = False
     component.run_timeout = 2 * 60 * 60
+    component.load_ml_calculating = False
     component.training_running = False
     component.training_cancelled = False
     return component
@@ -3754,8 +3755,11 @@ async def scenario():
         except asyncio.CancelledError:
             observed["cancelled"] = True
 
-        # The worker thread is still inside train_curriculum here. The next run() tick must not
-        # start a second one on the same predictor.
+        # The worker thread is still inside train_curriculum here. run()'s finally has already
+        # cleared load_ml_calculating, so is_calculating() has to get its answer from elsewhere,
+        # and the next run() tick must not start a second training on the same predictor.
+        component.load_ml_calculating = False
+        observed["calculating_while_orphaned"] = component.is_calculating()
         await component._do_training(is_initial=False)
         allow_exit.set()
 
@@ -3770,6 +3774,7 @@ async def scenario():
     assert component.training_cancelled is True, "a cancelled run must raise the component's own cancel signal, or the worker thread never learns to abandon"
     assert observed.get("hook_tripped") is True, "the stop hook must report True after the coroutine is cancelled, so the orphaned training thread abandons its run"
     assert observed["calls"] == 1, f"a second training must not start while the previous worker thread is still running, got {observed['calls']} calls"
+    assert observed.get("calculating_while_orphaned") is True, "is_calculating() must still report True while an orphaned training thread is doing NumPy work"
     assert component.training_running is False, "the worker thread must clear training_running when it returns, or training never runs again"
     assert component.last_train_time is None, "a cancelled run must not stamp last_train_time"
 

From 0d1ab6b2d55dcb0b66ccfe3bde0765c0926e74ec Mon Sep 17 00:00:00 2001
From: Rik Allen <rikallen@yahoo.com>
Date: Thu, 8 Oct 2026 10:39:29 +0100
Subject: [PATCH 4/5] Train a clone of the ML model and swap it in on success,
 instead of rolling back

Addresses the three open Copilot review threads on #5112, which all came from
training mutating the live LoadPredictor in place: an abandoned curriculum kept
the completed passes' weights, Adam state and normalisation behind its rolled-back
stamps, an aborted initial run left random weights marked model_initialized,
and an aborted train() kept the best epoch of the abandoned run rather than the
starting weights. An exception part way through had the same effect.

_do_training now trains LoadPredictor.clone() - an independent deep copy of
everything in TRAINING_STATE_FIELDS - and assigns it to self.predictor only
when the run succeeds. An abandoned, failed or raising run simply drops the
clone, so the live model is never part-trained and needs nothing rolled back.
That removes the piecemeal rollbacks: train()'s normalisation restore and
train_curriculum()'s stamp rollback. Called directly, an aborted train() or
curriculum now documents that it leaves the model part-trained.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 apps/predbat/load_ml_component.py  |  18 ++-
 apps/predbat/load_predictor.py     |  87 ++++++++------
 apps/predbat/tests/test_load_ml.py | 181 ++++++++++++++++-------------
 3 files changed, 167 insertions(+), 119 deletions(-)

diff --git a/apps/predbat/load_ml_component.py b/apps/predbat/load_ml_component.py
index 45d53d15e..b8ffaa380 100644
--- a/apps/predbat/load_ml_component.py
+++ b/apps/predbat/load_ml_component.py
@@ -988,9 +988,9 @@ async def _do_training(self, is_initial):
         if self.training_running:
             # A previous run()'s worker thread is still inside train_curriculum. That happens when
             # the run_timeout watchdog cancelled the coroutine awaiting it: the coroutine went away,
-            # the thread did not. Starting a second curriculum here would have two threads mutating
-            # the same LoadPredictor's weights, biases, Adam moments and normalisation statistics in
-            # place, and both racing to save() over the same model file.
+            # the thread did not. It is training its own clone, so the live predictor is safe, but a
+            # second curriculum would compete with it for the CPU, and resetting training_cancelled
+            # below would tell the orphan to carry on.
             self.log("Warn: ML Component: Previous training run is still finishing, skipping this training cycle")
             return
 
@@ -1029,6 +1029,12 @@ async def _do_training(self, is_initial):
                 max_intermediate_passes = self.ml_curriculum_max_passes
         # Lock released
 
+        # Train a clone, not the live predictor. Training mutates the model in place for the whole
+        # run, so the live one would be part-trained and racing the thread for anything that reads
+        # it in the meantime, and left part-trained by a run that is abandoned, fails or raises.
+        # The clone replaces it only once a run succeeds; otherwise it is simply dropped.
+        trainee = self.predictor.clone()
+
         # Tracks whether the stop hook ever actually answered True during this run. train() and
         # train_curriculum() both return None for "aborted" and for "every pass failed", so reading
         # the hook again after the fact would file a genuine data failure as an abandonment whenever
@@ -1050,7 +1056,7 @@ def run_training():
             """
             self.training_running = True
             try:
-                return self.predictor.train_curriculum(
+                return trainee.train_curriculum(
                     load_data_snap,
                     now_utc_snap,
                     pv_minutes=pv_data_snap,
@@ -1088,11 +1094,13 @@ def run_training():
                 raise
 
             if val_mae is None and stop_signalled["tripped"]:
-                # Abandoned on purpose, not a failure - and nothing to save, the model is partial
+                # Abandoned on purpose, not a failure. The part-trained clone is dropped, and the
+                # live model is exactly as it was before the run
                 self.log("ML Component: Training abandoned because the run was cancelled or the component is stopping")
                 return
 
             if val_mae is not None:
+                self.predictor = trainee
                 self.last_train_time = datetime.now(timezone.utc)
                 self.initial_training_done = True
 
diff --git a/apps/predbat/load_predictor.py b/apps/predbat/load_predictor.py
index 5f0c9343a..524791347 100644
--- a/apps/predbat/load_predictor.py
+++ b/apps/predbat/load_predictor.py
@@ -20,6 +20,7 @@
 and cyclical time features (minute-of-day, day-of-week, day-of-year).
 """
 
+import copy
 import numpy as np
 import json
 import os
@@ -291,6 +292,45 @@ def _reset_adam_optimizer(self):
         self.adam_t = 0
         self.log("ML Predictor: Reset Adam optimiser state for fine-tuning")
 
+    # Everything a training run changes. Weights, optimiser state, normalisation and the published
+    # stamps are one model, so clone() copies all of them rather than sharing any with the original.
+    TRAINING_STATE_FIELDS = (
+        "weights",
+        "biases",
+        "m_weights",
+        "v_weights",
+        "m_biases",
+        "v_biases",
+        "adam_t",
+        "feature_mean",
+        "feature_std",
+        "target_mean",
+        "target_std",
+        "model_initialized",
+        "model_trained",
+        "training_timestamp",
+        "validation_mae",
+        "validation_bias",
+        "rollout_mae",
+        "pattern_mae",
+        "epochs_trained",
+    )
+
+    def clone(self):
+        """Return an independent copy of this model to train while the original keeps serving.
+
+        Training mutates the model in place for minutes or hours, and an aborted, failed or raising
+        run leaves it part-trained. Training a clone and swapping it in only on success keeps the
+        live model whole and usable throughout, with nothing to roll back. The training state is
+        deep-copied because _adam_update changes the arrays in place; everything else - the
+        hyperparameters and log_func - is shared, as copy.deepcopy would also copy whatever object
+        log_func is bound to.
+        """
+        twin = copy.copy(self)
+        for field in self.TRAINING_STATE_FIELDS:
+            setattr(twin, field, copy.deepcopy(getattr(self, field)))
+        return twin
+
     def _forward(self, X, training=False):
         """
         Forward pass through the network.
@@ -1274,15 +1314,6 @@ def train(
             self.log("Warn: ML Predictor: No validation data available")
             return None
 
-        # Snapshot the normalisation statistics before they are refitted below. An aborted run
-        # restores its starting weights, but weights and normalisation are one model: leaving the
-        # newly fitted statistics in place alongside restored weights would have every subsequent
-        # predict() scale its inputs by statistics those weights were never trained against.
-        entry_feature_mean = self.feature_mean.copy() if self.feature_mean is not None else None
-        entry_feature_std = self.feature_std.copy() if self.feature_std is not None else None
-        entry_target_mean = self.target_mean
-        entry_target_std = self.target_std
-
         # Normalize features and targets
         # On initial train: fit normalization from scratch
         # On fine-tune: apply EMA update to track distribution drift gradually
@@ -1457,6 +1488,13 @@ def train(
                 self.log("ML Predictor: Early stopping at epoch {}".format(epoch + 1))
                 break
 
+        if aborted:
+            # Report failure so the caller neither publishes nor saves a partial model, and skip the
+            # AR rollout diagnostic, which is itself minutes of the work we just abandoned. The model
+            # is left part-trained - weights, optimiser state and normalisation are not rolled back -
+            # so a caller that needs its live model intact trains a clone() and discards it.
+            return None
+
         # Restore best weights after early stopping
         if best_weights is not None and best_biases is not None:
             self.weights = best_weights
@@ -1467,18 +1505,6 @@ def train(
                 )
             )
 
-        if aborted:
-            # Weights are restored above; put the normalisation statistics back with them so the
-            # in-memory model is left wholly as it was found, rather than as restored weights paired
-            # with statistics refitted for the training that was abandoned. Report failure so the
-            # caller neither publishes nor saves a partial model, and skip the AR rollout diagnostic,
-            # which is itself minutes of the work we just abandoned.
-            self.feature_mean = entry_feature_mean
-            self.feature_std = entry_feature_std
-            self.target_mean = entry_target_mean
-            self.target_std = entry_target_std
-            return None
-
         self.model_trained = True
         self.training_timestamp = datetime.now(timezone.utc)
         self.validation_mae = best_val_loss
@@ -1609,24 +1635,17 @@ def train_curriculum(
 
                 Returns:
                     Validation MAE from the final pass, or None if all passes failed or the run was
-                    aborted via stop_callback.
+                    aborted via stop_callback. An aborted curriculum leaves the model part-trained,
+                    with the completed passes' weights and stamps in place - train a clone() to keep
+                    a live model intact.
         """
         if self._should_stop(stop_callback):
             self.log("ML Predictor: Curriculum training not started - stop requested")
             return None
 
-        # Snapshot the published training stamps. train() stamps these at the end of EVERY completed
-        # pass, including the intermediate curriculum windows, so an abandoned curriculum would
-        # otherwise leave the last completed intermediate pass's timestamp and validation_mae behind
-        # - and is_valid() judges purely on those, so a model trained on the first 7 days of a 28-day
-        # curriculum would report "active" and its fresh timestamp would suppress the staleness
-        # retrain that is the only route back to a complete model.
-        entry_stamps = (self.model_trained, self.training_timestamp, self.validation_mae, self.validation_bias, self.epochs_trained)
-
         def abandon_curriculum(message):
-            """Log an abandoned curriculum, roll the published stamps back, and report no result."""
+            """Log an abandoned curriculum and report no result."""
             self.log(message)
-            self.model_trained, self.training_timestamp, self.validation_mae, self.validation_bias, self.epochs_trained = entry_stamps
             return None
 
         # Build list of positive minute keys to find total history span
@@ -2169,8 +2188,8 @@ def is_valid(self, validation_threshold=2.0, max_age_hours=48):
         if not self.model_trained:
             # model_initialized only means _initialize_weights() has run, which it does before the
             # first epoch, so weights alone can still be the random He initialisation. Without this
-            # an initial run abandoned mid-curriculum - whose stamps train_curriculum rolls back -
-            # would report "active" on a model that has never been trained. Tested separately from
+            # an initial run abandoned or failing before any pass completes would report "active"
+            # on a model that has never been trained. Tested separately from
             # training_timestamp, which a model saved before timestamps existed legitimately lacks.
             return False, "not_trained"
 
diff --git a/apps/predbat/tests/test_load_ml.py b/apps/predbat/tests/test_load_ml.py
index cc7fcb833..877393a33 100644
--- a/apps/predbat/tests/test_load_ml.py
+++ b/apps/predbat/tests/test_load_ml.py
@@ -10,6 +10,7 @@
 # fmt: on
 
 from utils import dp4
+import copy
 import numpy as np
 from datetime import datetime, timezone, timedelta
 import tempfile
@@ -59,8 +60,8 @@ def test_load_ml(my_predbat=None):
         ("training_stop_callback", _test_training_stops_on_stop_callback, "Training abandons a run when the stop hook trips"),
         ("training_off_event_loop", _test_component_training_does_not_block_event_loop, "Component training leaves its event loop free to run"),
         ("training_abandoned_on_stop", _test_component_training_abandons_run_on_stop, "Component abandons an in-flight training run when stopping"),
-        ("training_abort_restores_norm", _test_training_abort_restores_normalisation, "An aborted train() restores normalisation statistics with the weights"),
-        ("curriculum_mid_abort_rollback", _test_curriculum_mid_abort_rolls_back_stamps, "An abandoned curriculum rolls back the completed passes' published stamps"),
+        ("clone_is_independent", _test_clone_is_independent, "Training a clone leaves the original model untouched"),
+        ("training_swaps_clone_on_success", _test_component_training_swaps_in_clone_only_on_success, "Component swaps in its trained clone only when the run succeeds"),
         ("untrained_model_not_valid", _test_untrained_model_is_not_valid, "A model that never completed a training does not read as valid"),
         ("training_watchdog_cancel", _test_component_training_survives_watchdog_cancel, "A cancelled run() stops the training thread instead of orphaning it"),
         ("training_failure_not_abandon", _test_component_training_failure_is_not_reported_as_abandoned, "An all-passes-failed run is reported as a failure, not an abandonment"),
@@ -3382,6 +3383,14 @@ def bad_callback():
     return True
 
 
+class _SelfCloningFake:
+    """Base for fake predictors: clone() hands back the fake itself, so a test sees the calls _do_training makes."""
+
+    def clone(self):
+        """Return this fake, standing in for the independent copy a real predictor would return."""
+        return self
+
+
 def _make_training_component(predictor, now_utc):
     """Build a bare LoadMLComponent wired up with just the attributes _do_training reads."""
     import asyncio
@@ -3448,7 +3457,7 @@ def _test_component_marks_itself_alive_during_training():
     now_utc = datetime.now(timezone.utc)
     recorded = {}
 
-    class FakePredictor:
+    class FakePredictor(_SelfCloningFake):
         """Captures the kwargs the component passes, and pulses the callback like a real run."""
 
         def train_curriculum(self, *args, **kwargs):
@@ -3505,7 +3514,7 @@ def _test_component_training_does_not_block_event_loop():
     loop_released = [threading.Event(), threading.Event()]
     observed = {}
 
-    class BlockingPredictor:
+    class BlockingPredictor(_SelfCloningFake):
         """Blocks inside training the way a real curriculum pass does."""
 
         def train_curriculum(self, *args, **kwargs):
@@ -3561,7 +3570,7 @@ def _test_component_training_abandons_run_on_stop():
     now_utc = datetime.now(timezone.utc)
     observed = {}
 
-    class StoppablePredictor:
+    class StoppablePredictor(_SelfCloningFake):
         """Polls the stop hook the way a real pass does, and abandons the run once it trips."""
 
         def train_curriculum(self, *args, **kwargs):
@@ -3588,95 +3597,107 @@ def train_curriculum(self, *args, **kwargs):
     return True
 
 
-def _test_training_abort_restores_normalisation():
-    """An aborted train() must put back the normalisation statistics along with the weights.
+def _test_clone_is_independent():
+    """Training a clone must leave the original model exactly as it was.
 
-    Weights and normalisation are one model. train() refits the statistics before the first epoch
-    check, so restoring only best_weights would leave restored weights paired with statistics fitted
-    for the run that was abandoned, and every subsequent predict() would scale its inputs by numbers
-    those weights were never trained against. Only reachable while the component survives the abort
-    - a cancelled run(), not a process shutdown - which is exactly the run_timeout watchdog path.
+    The component trains a clone so the live model keeps serving, whole, while a run is in flight
+    and survives an abandoned, failed or raising run untouched. That only holds if no array is
+    shared: _adam_update changes the weights and moments in place, so a shallow copy would train
+    the live model too.
     """
     now_utc = datetime.now(timezone.utc)
     np.random.seed(11)
     load_data = _create_synthetic_load_data(n_days=7, now_utc=now_utc)
 
     predictor = LoadPredictor(learning_rate=0.01)
-    assert predictor.train(load_data, now_utc, epochs=2, patience=3) is not None, "the baseline run must succeed so there is a fitted model to abort against"
-
-    before = {
-        "feature_mean": predictor.feature_mean.copy(),
-        "feature_std": predictor.feature_std.copy(),
-        "target_mean": predictor.target_mean,
-        "target_std": predictor.target_std,
-        "weights": [w.copy() for w in predictor.weights],
-    }
-
-    # Fine-tune on plainly different data so the EMA refit moves the statistics by an amount no
-    # tolerance could absorb, then abort at the very first epoch check - after the refit has
-    # already happened
-    other_data = {minute: value * 5.0 for minute, value in load_data.items()}
-    aborted_mae = predictor.train(other_data, now_utc, epochs=5, patience=3, stop_callback=lambda: True)
-
-    assert aborted_mae is None, "an aborted fine-tune must report None"
-    assert np.allclose(predictor.feature_mean, before["feature_mean"]), "an aborted run must restore feature_mean, or predictions are normalised by statistics the restored weights never saw"
-    assert np.allclose(predictor.feature_std, before["feature_std"]), "an aborted run must restore feature_std"
-    assert predictor.target_mean == before["target_mean"], "an aborted run must restore target_mean"
-    assert predictor.target_std == before["target_std"], "an aborted run must restore target_std"
-    for n, (restored, original) in enumerate(zip(predictor.weights, before["weights"])):
-        assert np.allclose(restored, original), f"an aborted run must restore layer {n} weights unchanged"
+    assert predictor.train(load_data, now_utc, epochs=2, patience=3) is not None, "the baseline run must succeed so there is a model to clone"
+    before = {field: copy.deepcopy(getattr(predictor, field)) for field in LoadPredictor.TRAINING_STATE_FIELDS}
+
+    trainee = predictor.clone()
+    assert trainee is not predictor, "clone() must return a new object"
+    assert trainee.log == predictor.log, "the clone must log through the same function"
+    other_data = {minute: value * 3.0 for minute, value in load_data.items()}
+    assert trainee.train(other_data, now_utc, is_initial=False, epochs=2, patience=3) is not None, "the clone must be trainable"
+
+    for field, original in before.items():
+        current = getattr(predictor, field)
+        if isinstance(original, list):
+            for n, (now_value, then_value) in enumerate(zip(current, original)):
+                assert np.array_equal(now_value, then_value), f"training the clone changed the original's {field}[{n}]"
+        elif isinstance(original, np.ndarray):
+            assert np.array_equal(current, original), f"training the clone changed the original's {field}"
+        else:
+            assert current == original, f"training the clone changed the original's {field}: {original} -> {current}"
+    assert not all(np.array_equal(a, b) for a, b in zip(trainee.weights, predictor.weights)), "the clone's weights should have moved during its own training"
 
     return True
 
 
-def _test_curriculum_mid_abort_rolls_back_stamps():
-    """A curriculum abandoned mid-run must not leave a completed pass's stamps published.
+def _test_component_training_swaps_in_clone_only_on_success():
+    """_do_training must replace the live predictor only with a clone whose run succeeded.
 
-    train() stamps training_timestamp/validation_mae/epochs_trained at the end of every pass,
-    including the intermediate curriculum windows. is_valid() judges purely on those two, so without
-    a rollback a model trained on only the first window of the curriculum reports "active" and its
-    fresh timestamp suppresses the staleness retrain that is the only route back to a full model.
+    An abandoned run must leave the live model as it was - same object, same weights - rather than
+    part-trained, and so must a run that raises. A successful run must swap the trained clone in.
     """
+    import asyncio
+
     now_utc = datetime.now(timezone.utc)
     np.random.seed(13)
-    load_data = _create_synthetic_load_data(n_days=28, now_utc=now_utc)
+    load_data = _create_synthetic_load_data(n_days=7, now_utc=now_utc)
 
-    stale_timestamp = now_utc - timedelta(hours=100)
-    predictor = LoadPredictor(learning_rate=0.01)
-    predictor.model_initialized = True
-    predictor.model_trained = True
-    predictor.weights = [np.zeros((1, 1), dtype=np.float32)]
-    predictor.training_timestamp = stale_timestamp
-    predictor.validation_mae = 0.4
-    predictor.validation_bias = 0.05
-    predictor.epochs_trained = 42
-
-    assert predictor.is_valid(validation_threshold=2.0, max_age_hours=48) == (False, "stale"), "the starting model must be due a retrain, so the abandonment cannot be mistaken for it"
-
-    passes = {"count": 0}
-
-    def fake_train(*args, **kwargs):
-        """Stand in for a completed pass, stamping the model the way the real train() does."""
-        passes["count"] += 1
-        predictor.model_trained = True
-        predictor.training_timestamp = now_utc
-        predictor.validation_mae = 0.3
-        predictor.validation_bias = 0.01
-        predictor.epochs_trained += 3
-        return 0.3
-
-    predictor.train = fake_train
-    # Trips only once a pass has completed, so the run is abandoned between passes rather than
-    # before the first one - the path the existing stop-callback test never reaches
-    curriculum_mae = predictor.train_curriculum(load_data, now_utc, epochs=3, patience=3, curriculum_window_days=7, curriculum_step_days=7, stop_callback=lambda: passes["count"] >= 1)
-
-    assert passes["count"] == 1, f"exactly one pass should complete before the hook trips, got {passes['count']}"
-    assert curriculum_mae is None, "an abandoned curriculum must not return the completed intermediate pass's val_mae as a finished result"
-    assert predictor.training_timestamp == stale_timestamp, "an abandoned curriculum must roll back training_timestamp, or a partial model's fresh stamp suppresses the staleness retrain"
-    assert predictor.validation_mae == 0.4, "an abandoned curriculum must roll back validation_mae"
-    assert predictor.validation_bias == 0.05, "an abandoned curriculum must roll back validation_bias"
-    assert predictor.epochs_trained == 42, "an abandoned curriculum must roll back epochs_trained"
-    assert predictor.is_valid(validation_threshold=2.0, max_age_hours=48) == (False, "stale"), "the model must still read as due a retrain after an abandoned curriculum"
+    live = LoadPredictor(learning_rate=0.01)
+    assert live.train(load_data, now_utc, epochs=2, patience=3) is not None, "the baseline run must succeed so there is a live model"
+    live_weights = [w.copy() for w in live.weights]
+
+    component = _make_training_component(live, now_utc)
+    component.load_data = load_data
+    component.load_data_age_days = 7
+    component.ml_epochs_update = 3
+    logged = []
+    component.log = logged.append
+
+    # Abandoned: a stop lands after the first epoch
+    def stop_after_first_epoch():
+        """Stand in for a shutdown arriving mid-run."""
+        component.api_stop = True
+
+    component.update_success_timestamp = stop_after_first_epoch
+    asyncio.run(component._do_training(is_initial=False))
+    assert component.predictor is live, "an abandoned run must not swap its part-trained clone in"
+    for n, (now_value, then_value) in enumerate(zip(live.weights, live_weights)):
+        assert np.array_equal(now_value, then_value), f"an abandoned run must leave the live model's layer {n} weights untouched"
+    assert component.last_train_time is None, "an abandoned run must not stamp last_train_time"
+
+    # Raising: the trainee fails part way, after the live model has been cloned
+    class RaisingTrainee:
+        """A clone whose training blows up."""
+
+        def train_curriculum(self, *args, **kwargs):
+            """Fail the way a NumPy error mid-epoch would."""
+            raise FloatingPointError("simulated failure mid-epoch")
+
+    class RaisingPredictor:
+        """A live model whose clone fails to train."""
+
+        def clone(self):
+            """Hand back a trainee that raises."""
+            return RaisingTrainee()
+
+    raising_live = RaisingPredictor()
+    component.predictor = raising_live
+    component.api_stop = False
+    asyncio.run(component._do_training(is_initial=False))
+    assert component.predictor is raising_live, "a run that raises must not replace the live model"
+    assert any("Training exception" in msg for msg in logged), "a raising run must be logged as an exception"
+
+    # Successful: the trained clone replaces the live model, which itself was never touched
+    component.predictor = live
+    component.update_success_timestamp = lambda: None
+    asyncio.run(component._do_training(is_initial=False))
+    assert component.predictor is not live, "a successful run must swap its trained clone in"
+    assert component.last_train_time is not None, "a successful run must stamp last_train_time"
+    for n, (now_value, then_value) in enumerate(zip(live.weights, live_weights)):
+        assert np.array_equal(now_value, then_value), f"training must never touch the live model's layer {n} weights, only its clone's"
 
     return True
 
@@ -3722,7 +3743,7 @@ def _test_component_training_survives_watchdog_cancel():
     allow_exit = threading.Event()
     observed = {"calls": 0}
 
-    class CancelWatchPredictor:
+    class CancelWatchPredictor(_SelfCloningFake):
         """Polls the stop hook until it trips, then holds the thread open so the guard can be seen."""
 
         def train_curriculum(self, *args, **kwargs):
@@ -3794,7 +3815,7 @@ def _test_component_training_failure_is_not_reported_as_abandoned():
     now_utc = datetime.now(timezone.utc)
     logged = []
 
-    class FailingPredictor:
+    class FailingPredictor(_SelfCloningFake):
         """Fails every pass without the stop hook ever tripping, then a stop lands at the end."""
 
         def train_curriculum(self, *args, **kwargs):

From 51f121e57d1eca98228b23191a4d63feb04af234 Mon Sep 17 00:00:00 2001
From: Rik Allen <rikallen@yahoo.com>
Date: Thu, 8 Oct 2026 11:03:53 +0100
Subject: [PATCH 5/5] Run ML training as a background task, judged on progress
 rather than run_timeout

Moving training off the event loop woke up ComponentBase.start()'s run_timeout
watchdog, which could never fire while training pinned the loop. load_ml's is
2 hours, so a first curriculum longer than that on slow hardware would be
cancelled - and its clone thrown away - every time, and that model would never
exist.

run() now launches training with _start_training as a task on the component's
own loop and returns, so the watchdog never covers it. Because the run trains a
clone, later ticks keep fetching and predicting from the live model; they only
hold off a second run. The tick after a run finishes re-fetches and publishes,
replacing the post-training stale re-fetch, and the clone is taken on the
worker thread inside the error handling.

_watch_training judges the run on progress (time.monotonic(), so a host
suspend is not read as a stall): no epoch for 2 hours logs an Error and asks
the run to abandon itself, and a run still progressing after 2 hours is logged
once. A failed, abandoned or raising run is retried after 30 minutes rather
than relaunched every minute. sensor.predbat_load_ml_stats gains
training_status (idle/training/stalled), training_started,
training_last_progress and training_retry_after, also published during a first
run, and docs/components.md describes them.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 apps/predbat/load_ml_component.py  | 249 ++++++++++++++++-----
 apps/predbat/load_predictor.py     |  10 +-
 apps/predbat/tests/test_load_ml.py | 341 ++++++++++++++++++++++++++---
 docs/components.md                 |   5 +
 4 files changed, 510 insertions(+), 95 deletions(-)

diff --git a/apps/predbat/load_ml_component.py b/apps/predbat/load_ml_component.py
index b8ffaa380..c812a29e1 100644
--- a/apps/predbat/load_ml_component.py
+++ b/apps/predbat/load_ml_component.py
@@ -21,6 +21,7 @@
 
 import asyncio
 import os
+import time
 from datetime import datetime, timezone, timedelta
 from component_base import ComponentBase
 from utils import get_now_from_cumulative, dp2, dp4, minute_data
@@ -34,6 +35,17 @@
 RETRAIN_INTERVAL_SECONDS = 2 * 60 * 60  # 2 hours between training cycles
 PREDICTION_INTERVAL_SECONDS = 30 * 60  # 30 minutes between predictions
 
+# Training runs as a background task outside run()'s run_timeout, because a first curriculum on
+# slow hardware can legitimately take longer than any fixed cap - and a cancelled run is thrown
+# away, so a cap it always exceeds means that model never exists. It is judged on progress instead:
+# no completed epoch for this long means it is stuck.
+TRAINING_STALL_SECONDS = 2 * 60 * 60
+# A run still progressing after this long is logged once, so slow training is visible, not silent
+TRAINING_SLOW_WARN_SECONDS = 2 * 60 * 60
+# Wait after a run that failed or was abandoned before trying again, rather than relaunching the
+# same doomed run every minute
+TRAINING_RETRY_SECONDS = 30 * 60
+
 # Database schema version - increment when the saved format changes to force a clean rebuild
 DATABASE_VERSION = 1
 
@@ -95,7 +107,7 @@ def initialize(self, load_ml_enable, load_ml_source=True, load_ml_max_days_histo
         self.ml_curriculum_max_passes = 4  # Max intermediate passes (0 = no limit)
         self.load_ml_database_days = load_ml_database_days
         self.ml_validation_holdout_hours = 48
-        self.run_timeout = 2 * 60 * 60  # 2 hours timeout for the whole run to prevent runaway execution
+        self.run_timeout = 2 * 60 * 60  # 2 hours timeout for one run() - training is not under it, see TRAINING_STALL_SECONDS
 
         # Data state
         self.load_data = None
@@ -121,12 +133,21 @@ def initialize(self, load_ml_enable, load_ml_source=True, load_ml_max_days_histo
         self.initial_training_done = False
         self.database_history_loaded = False
 
-        # In-flight training state. training_running is owned by the worker thread (it clears it
-        # when the synchronous curriculum truly returns, not when the awaiting coroutine goes away),
-        # and training_cancelled is the component's own cancel signal for a run whose coroutine has
-        # been cancelled out from under it - see _do_training.
+        # In-flight training state. training_task is the background task run() launches training as.
+        # training_running is owned by the worker thread (it clears it when the synchronous
+        # curriculum truly returns, not when the awaiting task goes away), and training_cancelled is
+        # the component's own cancel signal for a run that has been given up on - see _do_training.
+        # The time.monotonic() stamps and flags feed _watch_training's stall and slow-run checks -
+        # monotonic so a host suspend or clock step is not read as a stall - and
+        # training_retry_after holds off a new run after one fails.
+        self.training_task = None
         self.training_running = False
         self.training_cancelled = False
+        self.training_started_monotonic = None
+        self.training_progress_monotonic = None
+        self.training_stalled = False
+        self.training_slow_warned = False
+        self.training_retry_after = None
 
         # Predictions cache
         self.current_predictions = {}
@@ -179,12 +200,27 @@ def _init_predictor(self):
     def is_calculating(self):
         """Return whether the component is currently doing NumPy-heavy work.
 
-        training_running is part of the answer because a training run outlives the coroutine that
-        started it: when the run_timeout watchdog cancels run(), its finally clears
-        load_ml_calculating while the worker thread is still inside train_curriculum.
+        training_running is part of the answer because training runs in a background thread long
+        after the run() that launched it has cleared load_ml_calculating.
         """
         return self.load_ml_calculating or self.training_running
 
+    def training_in_flight(self):
+        """Return True while a training run's task or worker thread has not finished.
+
+        Both, because either can outlive the other: the thread keeps going after its task is
+        cancelled at shutdown, and the task has work of its own before and after the thread.
+        """
+        return self.training_running or (self.training_task is not None and not self.training_task.done())
+
+    def training_status(self):
+        """Return idle, training or stalled, for the stats sensor."""
+        if self.training_stalled:
+            return "stalled"
+        if self.training_in_flight():
+            return "training"
+        return "idle"
+
     def get_from_incrementing(self, data, index, step, backwards=True):
         """
         Get a single value from an incrementing series e.g. kWh today -> kWh this minute
@@ -729,15 +765,28 @@ async def run(self, seconds, first):
             self.api_started = True
             return True
 
+        # A run launched by an earlier tick may have finished, swapping in a new model, or still be
+        # going. It trains a clone, so this tick's own fetch and prediction are free to proceed.
+        training_finished = self._collect_training()
+        training_in_flight = self.training_in_flight()
+        stall_flagged = self._watch_training() if training_in_flight else False
+        if not training_in_flight:
+            # Cleared here rather than when the task is collected: a thread can outlive its task
+            self.training_stalled = False
+        retry_wait = self.training_retry_after is not None and self.now_utc < self.training_retry_after
+
         # Determine if training is needed
         is_initial = not self.initial_training_done
 
         # Retrain if the model is older than the retrain interval (rather than on a fixed tick)
         retrain_age_seconds = (self.now_utc - self.last_train_time).total_seconds() if self.last_train_time else RETRAIN_INTERVAL_SECONDS
-        should_train = not first and (retrain_age_seconds >= RETRAIN_INTERVAL_SECONDS)
+        should_train = not first and not training_in_flight and not retry_wait and (retrain_age_seconds >= RETRAIN_INTERVAL_SECONDS)
 
-        # Fetch fresh load data periodically (every N minutes)
-        should_fetch = first or should_train or ((seconds % PREDICTION_INTERVAL_SECONDS) == 0)
+        # Fetch fresh load data periodically (every N minutes), and straight after a training run
+        # lands so its model is published on current data rather than data fetched before a run
+        # that may have taken hours - one that crossed midnight would otherwise publish yesterday's
+        # daily total as today's baseline.
+        should_fetch = first or should_train or training_finished or stall_flagged or ((seconds % PREDICTION_INTERVAL_SECONDS) == 0)
 
         # Load database
         if not self.database_history_loaded and self.load_ml_database_days:
@@ -764,6 +813,15 @@ async def run(self, seconds, first):
         if is_initial:
             if should_train:
                 self.log("ML Component: Starting initial training")
+            elif training_in_flight or retry_wait:
+                # Nothing to predict with until the first run lands, but the component itself is
+                # working: the run is watched by _watch_training and a failure has logged its retry.
+                # The stats sensor still carries the run's progress, a stall or the retry time.
+                if should_fetch:
+                    self._update_model_status()
+                    self._publish_entity()
+                self.update_success_timestamp()
+                return True
             else:
                 self.log("ML Component: Initial training is required, delaying until component has started")
                 return True
@@ -771,7 +829,12 @@ async def run(self, seconds, first):
             self.log("ML Component: Starting fine-tune training (2h interval), model age is {} hours".format(retrain_age_seconds / 3600.0))
         elif should_fetch:
             # If not training either then no need to print anything
-            self.log("ML Component: No training needed, model age is {} hours".format(dp2(retrain_age_seconds / 3600.0)))
+            if training_in_flight:
+                self.log("ML Component: Training in progress, model age is {} hours".format(dp2(retrain_age_seconds / 3600.0)))
+            elif retry_wait:
+                self.log("ML Component: Training will be retried after {}, model age is {} hours".format(self.training_retry_after.strftime(TIME_FORMAT), dp2(retrain_age_seconds / 3600.0)))
+            else:
+                self.log("ML Component: No training needed, model age is {} hours".format(dp2(retrain_age_seconds / 3600.0)))
 
         if should_train or should_fetch:
             # Set load_ml_calculating across all NumPy-heavy work (training + predict + save). This used
@@ -788,28 +851,15 @@ async def run(self, seconds, first):
             try:
                 self.load_ml_calculating = True
 
-                if should_train and not self.api_stop:
-                    self.log("ML Component: Doing training...")
-                    await self._do_training(is_initial)
-
                 if self.api_stop:
-                    # A re-fetch and a prediction cycle here would just add their own delay to the
-                    # stop we are already answering. The model status below still runs: training may
-                    # have completed successfully before the stop landed, and skipping it entirely
-                    # would leave that fresh model unpublished.
-                    self.log("ML Component: Stopping, skipping the post-training prediction cycle")
+                    # A prediction cycle or a new training run would just add their own delay to the
+                    # stop we are already answering
+                    self.log("ML Component: Stopping, skipping this cycle's ML work")
                     should_fetch = False
-                elif should_train:
-                    # Training can run for many minutes, during which the data fetched above
-                    # goes stale - both the lookback window feeding the prediction and the
-                    # load_minutes_now baseline still refer to the pre-training time. Re-fetch
-                    # so everything is anchored to the current time again. Without this a
-                    # training run that straddles midnight publishes yesterday's daily total
-                    # as today's baseline, inflating the forecast by a whole day of load.
-                    stale_minutes = (self.now_utc - self.last_data_fetch).total_seconds() / 60.0 if self.last_data_fetch else 0
-                    if stale_minutes >= PREDICT_STEP:
-                        self.log("ML Component: Data is {:.0f} minutes stale after training, re-fetching before prediction".format(stale_minutes))
-                        await self._do_fetch()
+                    should_train = False
+
+                if should_train:
+                    self._start_training(is_initial)
 
                 # Update model validity status
                 self._update_model_status()
@@ -818,7 +868,7 @@ async def run(self, seconds, first):
                     # Get predictions for the current cycle (will be used by fetch.py to publish forecast entity)
                     self._get_predictions(self.now_utc, self.midnight_utc)
 
-                    # Publish entity with current state
+                    # Publish entity with current state, including whether training is running
                     self._publish_entity()
                     self.log("ML Component: Prediction cycle completed")
 
@@ -971,29 +1021,95 @@ def array_to_dict(arr):
     def training_stop_requested(self):
         """Return True when this training run has been cancelled, so the worker thread abandons it.
 
-        Three sources, because a training run outlives its coroutine: api_stop for an ordinary
-        shutdown, fatal_error because ComponentBase.start() exits on it without ever calling stop(),
-        and training_cancelled for a run() the run_timeout watchdog has cancelled - in that last case
-        api_stop is still False, so without this the worker thread would train on unwatched.
+        Three sources, because a training run outlives its task: api_stop for an ordinary shutdown,
+        fatal_error because ComponentBase.start() exits on it without ever calling stop(), and
+        training_cancelled for a run given up on while the component carries on - one that stalled,
+        or whose task was cancelled - where api_stop is still False, so without it the worker
+        thread would train on unwatched.
         """
         return self.api_stop or self.fatal_error or self.training_cancelled
 
+    def _start_training(self, is_initial):
+        """Launch a training run as a background task on this component's loop, without awaiting it.
+
+        Awaiting it would put the run under run()'s run_timeout, and the watchdog would cancel - and
+        so throw away - any curriculum longer than that, which a first run on slow hardware can
+        legitimately be, every time, so that model would never exist. It trains a clone, so later
+        ticks keep predicting from the live model meanwhile, and _watch_training judges it on
+        progress instead of a deadline.
+        """
+        self.training_started_monotonic = time.monotonic()
+        self.training_progress_monotonic = self.training_started_monotonic
+        self.training_stalled = False
+        self.training_slow_warned = False
+        self.training_retry_after = None
+        self.log("ML Component: Doing training...")
+        self.training_task = asyncio.ensure_future(self._do_training(is_initial))
+
+    def _training_progress(self):
+        """Per-epoch heartbeat from the worker thread, for _watch_training's stall check."""
+        self.training_progress_monotonic = time.monotonic()
+
+    def _watch_training(self):
+        """Check an in-flight training run for a stall, and log one that is slow but progressing.
+
+        A run with no completed epoch for TRAINING_STALL_SECONDS is told to abandon itself and
+        flagged on the stats sensor. The live model keeps serving either way. A thread stuck inside a
+        single epoch will not see the request until it gets out, so the flag stays up until it does.
+
+        Returns:
+            True when this call newly flagged a stall, so the caller can publish it
+        """
+        if self.training_stalled:
+            return False
+
+        now = time.monotonic()
+        silent_seconds = now - self.training_progress_monotonic if self.training_progress_monotonic is not None else 0
+        if silent_seconds >= TRAINING_STALL_SECONDS:
+            self.training_stalled = True
+            self.training_cancelled = True
+            self.log("Error: ML Component: Training has made no progress for {} minutes, asking it to abandon the run - the current model stays in use".format(int(silent_seconds // 60)))
+            return True
+
+        running_seconds = now - self.training_started_monotonic if self.training_started_monotonic is not None else 0
+        if running_seconds >= TRAINING_SLOW_WARN_SECONDS and not self.training_slow_warned:
+            self.training_slow_warned = True
+            self.log("Warn: ML Component: Training has been running for {:.1f} hours and is still progressing (last epoch {:.0f} minutes ago)".format(running_seconds / 3600.0, silent_seconds / 60.0))
+        return False
+
+    def _training_time_iso(self, monotonic_stamp):
+        """Render a time.monotonic() stamp from the in-flight run as a wall-clock ISO time, or None."""
+        if monotonic_stamp is None or not self.training_in_flight():
+            return None
+        return (datetime.now(timezone.utc) - timedelta(seconds=time.monotonic() - monotonic_stamp)).isoformat()
+
+    def _collect_training(self):
+        """Clear a finished training task, returning True when one has finished since the last tick."""
+        task = self.training_task
+        if task is None or not task.done():
+            return False
+        self.training_task = None
+        if not task.cancelled() and task.exception() is not None:
+            # _do_training catches its own exceptions, so this is a bug in that handling - report
+            # it rather than letting it vanish with the task
+            self.log("Error: ML Component: Training task failed: {}".format(task.exception()))
+        return True
+
+    def _schedule_training_retry(self):
+        """Hold off the next training run after one that failed or was abandoned."""
+        self.training_retry_after = self.now_utc + timedelta(seconds=TRAINING_RETRY_SECONDS)
+        self.log("ML Component: Will retry training after {}".format(self.training_retry_after.strftime(TIME_FORMAT)))
+
     async def _do_training(self, is_initial):
         """
-        Perform model training.
+        Perform one model training run, as the background task _start_training launches.
+
+        Trains a clone of the live predictor on a worker thread and swaps it in only if the run
+        succeeds, scheduling a retry if it fails or is abandoned.
 
         Args:
             is_initial: True for full training, False for fine-tuning
         """
-        if self.training_running:
-            # A previous run()'s worker thread is still inside train_curriculum. That happens when
-            # the run_timeout watchdog cancelled the coroutine awaiting it: the coroutine went away,
-            # the thread did not. It is training its own clone, so the live predictor is safe, but a
-            # second curriculum would compete with it for the CPU, and resetting training_cancelled
-            # below would tell the orphan to carry on.
-            self.log("Warn: ML Component: Previous training run is still finishing, skipping this training cycle")
-            return
-
         # Snapshot data under the lock so we can release it before the CPU-bound
         # training call.  A curriculum run is many passes of many epochs and routinely takes
         # minutes, so it must NOT run while holding data_lock, and must not run on the event
@@ -1001,6 +1117,7 @@ async def _do_training(self, is_initial):
         async with self.data_lock:
             if not self.load_data:
                 self.log("Warn: ML Component: No data for training")
+                self._schedule_training_retry()
                 return
 
             if self.load_data_age_days < 3:
@@ -1032,8 +1149,10 @@ async def _do_training(self, is_initial):
         # Train a clone, not the live predictor. Training mutates the model in place for the whole
         # run, so the live one would be part-trained and racing the thread for anything that reads
         # it in the meantime, and left part-trained by a run that is abandoned, fails or raises.
-        # The clone replaces it only once a run succeeds; otherwise it is simply dropped.
-        trainee = self.predictor.clone()
+        # The clone replaces it only once a run succeeds; otherwise it is simply dropped. It is
+        # taken on the worker thread, keeping the deep copy off the loop and inside the error
+        # handling below.
+        live_predictor = self.predictor
 
         # Tracks whether the stop hook ever actually answered True during this run. train() and
         # train_curriculum() both return None for "aborted" and for "every pass failed", so reading
@@ -1056,7 +1175,8 @@ def run_training():
             """
             self.training_running = True
             try:
-                return trainee.train_curriculum(
+                trainee = live_predictor.clone()
+                return trainee, trainee.train_curriculum(
                     load_data_snap,
                     now_utc_snap,
                     pv_minutes=pv_data_snap,
@@ -1070,7 +1190,7 @@ def run_training():
                     curriculum_window_days=window_days,
                     curriculum_step_days=step_days,
                     max_intermediate_passes=max_intermediate_passes,
-                    progress_callback=self.update_success_timestamp,
+                    progress_callback=self._training_progress,
                     stop_callback=stop_hook,
                 )
             finally:
@@ -1080,27 +1200,30 @@ def run_training():
         try:
             # Run the curriculum on a worker thread. It is synchronous, CPU-bound and minutes long,
             # so calling it directly would pin this component's event loop for its whole duration:
-            # api_stop would go unnoticed, the run_timeout watchdog could never fire, and a shutdown
-            # landing mid-training would sit waiting on this component's thread (#5075).
+            # api_stop would go unnoticed and a shutdown landing mid-training would sit waiting on
+            # this component's thread (#5075).
             try:
-                val_mae = await asyncio.to_thread(run_training)
+                trainee, val_mae = await asyncio.to_thread(run_training)
             except asyncio.CancelledError:
-                # Now that the loop is free the run_timeout watchdog can fire, and cancelling this
-                # coroutine does nothing to the worker thread already inside train_curriculum.
-                # Trip our own stop signal so the thread abandons itself at its next epoch check
-                # instead of training on unwatched, then let the cancellation propagate.
+                # Cancelling this task - the component's loop being torn down at shutdown, say -
+                # does nothing to the worker thread already inside train_curriculum. Trip our own
+                # stop signal so the thread abandons itself at its next epoch check instead of
+                # training on unwatched, then let the cancellation propagate.
                 self.training_cancelled = True
-                self.log("Warn: ML Component: Training run was cancelled (run_timeout is {}s) - asking the training thread to abandon the run".format(self.run_timeout))
+                self.log("Warn: ML Component: Training task was cancelled - asking the training thread to abandon the run")
                 raise
 
             if val_mae is None and stop_signalled["tripped"]:
                 # Abandoned on purpose, not a failure. The part-trained clone is dropped, and the
                 # live model is exactly as it was before the run
-                self.log("ML Component: Training abandoned because the run was cancelled or the component is stopping")
+                self.log("ML Component: Training abandoned because the run stalled, was cancelled or the component is stopping")
+                if not self.api_stop and not self.fatal_error:
+                    self._schedule_training_retry()
                 return
 
             if val_mae is not None:
                 self.predictor = trainee
+                self.training_retry_after = None
                 self.last_train_time = datetime.now(timezone.utc)
                 self.initial_training_done = True
 
@@ -1119,10 +1242,12 @@ def run_training():
                     self.predictor.save(self.model_filepath)
             else:
                 self.log("Warn: ML Component: Training failed")
+                self._schedule_training_retry()
 
         except Exception as e:
             self.log("Error: ML Component: Training exception: {}".format(e))
             self.log("Error: " + traceback.format_exc())
+            self._schedule_training_retry()
 
     def _update_model_status(self):
         """Update model validity status based on current state."""
@@ -1235,6 +1360,10 @@ def _publish_entity(self):
                 "model_age_hours": round(model_age_hours, 1) if model_age_hours is not None else None,
                 "training_days": self.load_data_age_days,
                 "status": self.model_status,
+                "training_status": self.training_status(),
+                "training_started": self._training_time_iso(self.training_started_monotonic),
+                "training_last_progress": self._training_time_iso(self.training_progress_monotonic),
+                "training_retry_after": self.training_retry_after.isoformat() if self.training_retry_after else None,
                 "model_version": MODEL_VERSION,
                 "epochs_trained": self.predictor.epochs_trained if self.predictor else 0,
                 "friendly_name": "ML Load Stats",
diff --git a/apps/predbat/load_predictor.py b/apps/predbat/load_predictor.py
index 524791347..1a85105b0 100644
--- a/apps/predbat/load_predictor.py
+++ b/apps/predbat/load_predictor.py
@@ -1279,9 +1279,9 @@ def train(
             huber_delta: Huber loss transition point in normalised target units (default 1.35; errors within
                          this threshold are penalised quadratically, beyond it linearly)
             ema_smoothing_alpha: EMA alpha for smoothing the early-stopping metric across epochs (0=no smoothing, 0.3=moderate)
-            progress_callback: Called with no arguments once per epoch, so a caller whose liveness is
-                               judged on how recently it reported success can stay alive through a
-                               training run longer than that window. Exceptions are logged and swallowed
+            progress_callback: Called with no arguments once per epoch, so a caller can tell a long
+                               run that is progressing from one that is stuck. Exceptions are logged
+                               and swallowed
             stop_callback: Called with no arguments at the top of every epoch; returning True abandons
                            the run. Distinct from progress_callback precisely because that hook must
                            never be able to abort a run - this one is the caller explicitly asking to.
@@ -1621,8 +1621,8 @@ def train_curriculum(
                     validation_holdout_hours: Holdout window for the final pass
                     norm_ema_alpha: Normalisation EMA alpha for passes after the first
                     progress_callback: Called with no arguments once per epoch across every pass, so a
-                                       run that lasts longer than the caller's liveness timeout can
-                                       still report itself alive
+                                       caller can tell a long run that is progressing from one that
+                                       is stuck
                     stop_callback: Called with no arguments before each pass and, via train(), before each
                                    epoch; returning True abandons the whole curriculum and returns None
                     curriculum_window_days: Initial training window size in days (default 7)
diff --git a/apps/predbat/tests/test_load_ml.py b/apps/predbat/tests/test_load_ml.py
index 877393a33..ae3503b82 100644
--- a/apps/predbat/tests/test_load_ml.py
+++ b/apps/predbat/tests/test_load_ml.py
@@ -56,14 +56,18 @@ def test_load_ml(my_predbat=None):
         ("fine_tune", _test_fine_tune, "Fine-tune on recent data"),
         ("curriculum_training", _test_curriculum_training, "Curriculum training with progressive window expansion"),
         ("training_progress_callback", _test_training_progress_callback, "Training reports liveness through progress_callback"),
-        ("component_alive_during_training", _test_component_marks_itself_alive_during_training, "Component keeps its success timestamp fresh while training"),
+        ("training_progress_reported", _test_component_records_training_progress, "Component records training progress from every epoch"),
         ("training_stop_callback", _test_training_stops_on_stop_callback, "Training abandons a run when the stop hook trips"),
         ("training_off_event_loop", _test_component_training_does_not_block_event_loop, "Component training leaves its event loop free to run"),
         ("training_abandoned_on_stop", _test_component_training_abandons_run_on_stop, "Component abandons an in-flight training run when stopping"),
         ("clone_is_independent", _test_clone_is_independent, "Training a clone leaves the original model untouched"),
         ("training_swaps_clone_on_success", _test_component_training_swaps_in_clone_only_on_success, "Component swaps in its trained clone only when the run succeeds"),
         ("untrained_model_not_valid", _test_untrained_model_is_not_valid, "A model that never completed a training does not read as valid"),
-        ("training_watchdog_cancel", _test_component_training_survives_watchdog_cancel, "A cancelled run() stops the training thread instead of orphaning it"),
+        ("training_task_cancel", _test_component_training_survives_task_cancel, "A cancelled training task stops the training thread instead of orphaning it"),
+        ("training_in_background", _test_component_training_runs_in_background, "run() trains in the background and keeps predicting meanwhile"),
+        ("initial_training_published", _test_component_initial_training_is_published, "Ticks during a first training run keep the stats sensor current"),
+        ("training_stall_and_slow", _test_component_training_stall_and_slow_run, "A slow training run is logged once and a stalled one abandoned and reported"),
+        ("training_retry_backoff", _test_component_training_retry_backoff, "A failed training run is retried after a backoff, not on the next tick"),
         ("training_failure_not_abandon", _test_component_training_failure_is_not_reported_as_abandoned, "An all-passes-failed run is reported as a failure, not an abandonment"),
         ("prediction", _test_prediction, "End-to-end prediction"),
         ("prediction_with_pv", _test_prediction_with_pv, "Prediction with PV forecast data"),
@@ -1120,6 +1124,8 @@ async def mock_save_database_history():
         result = await component.run(seconds=30, first=False)
 
         assert result is True, "Second run should return True"
+        assert component.training_task is not None, "Training should be launched as a background task on the second run"
+        await component.training_task
         assert training_call_count[0] == 1, f"Initial training should run on second run, ran {training_call_count[0]} times"
         assert component.initial_training_done is True, "initial_training_done should be True after second run"
         assert save_call_count[0] == 1, f"save_database_history should be called once after second run, called {save_call_count[0]} times"
@@ -1144,7 +1150,7 @@ async def mock_save_database_history():
         # last_train_time = 30 min ago (set in mock_do_training to component.now_utc of Run 2).
         # retrain_age_seconds = 30*60 = 1800 < RETRAIN_INTERVAL_SECONDS (7200) -> should_train=False.
         # seconds % PREDICTION_INTERVAL_SECONDS == 0 -> should_fetch=True.
-        # Expect: fetch+predict+save only, no training.
+        # Expect: fetch+predict+save only, no training (the finished run's task is collected).
         component._fetch_load_data = AsyncMock(return_value=(fetch_data_3, 7, 3.0, None, None, None, None))
 
         result = await component.run(seconds=PREDICTION_INTERVAL_SECONDS, first=False)
@@ -1152,6 +1158,7 @@ async def mock_save_database_history():
         assert result is True, "Third run should return True"
         assert training_call_count[0] == 1, f"Training should NOT run again on third run (model too fresh), ran {training_call_count[0]} times total"
         assert save_call_count[0] == 2, f"save_database_history should be called again on third run, called {save_call_count[0]} times"
+        assert component.training_task is None, "The finished training task should be collected by the next run"
 
         print("    PASS: Run 3: fetch-only cycle (model fresh), save called without retraining")
 
@@ -2753,9 +2760,9 @@ def mock_dashboard_item(entity_id, state, attributes, app):
 
     print("    PASS: Missing snapshot timestamp falls back to the stored baseline")
 
-    # Re-fetch after a long training run must re-anchor the baseline before predicting
+    # The tick after a long training run must re-fetch, re-anchoring the baseline before predicting
     async def run_stale_refetch():
-        """Drive run() through a training cycle that leaves the fetched data stale."""
+        """Drive run() through a training run that crosses midnight, then the tick that collects it."""
         fetch_times = []
 
         async def mock_fetch_load_data():
@@ -2769,6 +2776,7 @@ async def mock_do_training(is_initial):
             mock_base.now_utc = datetime(2026, 1, 2, 0, 15, 0, tzinfo=timezone.utc)
             mock_base.midnight_utc = datetime(2026, 1, 2, 0, 0, 0, tzinfo=timezone.utc)
             mock_base.minutes_now = 15
+            component.last_train_time = mock_base.now_utc
 
         component._fetch_load_data = mock_fetch_load_data
         component._do_training = mock_do_training
@@ -2789,6 +2797,10 @@ async def mock_do_training(is_initial):
         mock_base.dashboard_calls = []
 
         await component.run(0, False)
+        await component.training_task
+        mock_base.dashboard_calls = []
+        # Not a 30-minute boundary, so only the finished training run can make this tick fetch
+        await component.run(60, False)
 
         assert len(fetch_times) == 2, f"Expected a second fetch after training, got {len(fetch_times)} fetches"
         assert fetch_times[1].day == 2, f"Second fetch should happen after midnight, got {fetch_times[1]}"
@@ -3441,16 +3453,22 @@ def log(self, msg, **kwargs):
     component.load_ml_calculating = False
     component.training_running = False
     component.training_cancelled = False
+    component.training_task = None
+    component.training_started_monotonic = None
+    component.training_progress_monotonic = None
+    component.training_stalled = False
+    component.training_slow_warned = False
+    component.training_retry_after = None
     return component
 
 
-def _test_component_marks_itself_alive_during_training():
-    """_do_training must hand the trainer the component's own timestamp updater.
+def _test_component_records_training_progress():
+    """_do_training must hand the trainer the component's own progress recorder.
 
-    Driven through the real _do_training rather than by calling train_curriculum directly - a test
-    that passes the callback itself proves only that the test passes it, and would keep passing if
-    load_ml_component stopped wiring it up. That regression is invisible until a training run
-    silently crosses the liveness window again hours later.
+    _watch_training judges an in-flight run on how recently an epoch completed, so a run whose
+    epochs go unreported reads as stalled and is abandoned. Driven through the real _do_training
+    rather than by calling train_curriculum directly - a test that passes the callback itself proves
+    only that the test passes it, and would keep passing if load_ml_component stopped wiring it up.
     """
     import asyncio
 
@@ -3479,15 +3497,15 @@ def train_curriculum(self, *args, **kwargs):
     # and wiring only one of them up leaves the other able to blow past the liveness window unnoticed
     for is_initial in (False, True):
         recorded.clear()
-        component.last_success_timestamp = None
+        component.training_progress_monotonic = None
         asyncio.run(component._do_training(is_initial=is_initial))
 
         calls = recorded.get("calls", [])
         assert len(calls) == 1, f"is_initial={is_initial}: expected exactly 1 train_curriculum call, got {len(calls)}"
         for n, callback in enumerate(calls):
             assert callback is not None, f"_do_training(is_initial={is_initial}) call {n + 1} must pass a progress_callback to train_curriculum"
-            assert callback == component.update_success_timestamp, f"is_initial={is_initial} call {n + 1}: the callback must be the component's own update_success_timestamp, not some other hook"
-        assert component.last_success_timestamp is not None, f"is_initial={is_initial}: firing the callback must move last_success_timestamp so components.is_alive() sees a recent success"
+            assert callback == component._training_progress, f"is_initial={is_initial} call {n + 1}: the callback must be the component's own _training_progress, not some other hook"
+        assert component.training_progress_monotonic is not None, f"is_initial={is_initial}: firing the callback must record progress, or _watch_training flags a working run as stalled"
 
     return True
 
@@ -3661,7 +3679,7 @@ def stop_after_first_epoch():
         """Stand in for a shutdown arriving mid-run."""
         component.api_stop = True
 
-    component.update_success_timestamp = stop_after_first_epoch
+    component._training_progress = stop_after_first_epoch
     asyncio.run(component._do_training(is_initial=False))
     assert component.predictor is live, "an abandoned run must not swap its part-trained clone in"
     for n, (now_value, then_value) in enumerate(zip(live.weights, live_weights)):
@@ -3692,7 +3710,7 @@ def clone(self):
 
     # Successful: the trained clone replaces the live model, which itself was never touched
     component.predictor = live
-    component.update_success_timestamp = lambda: None
+    component._training_progress = lambda: None
     asyncio.run(component._do_training(is_initial=False))
     assert component.predictor is not live, "a successful run must swap its trained clone in"
     assert component.last_train_time is not None, "a successful run must stamp last_train_time"
@@ -3725,18 +3743,279 @@ def _test_untrained_model_is_not_valid():
     return True
 
 
-def _test_component_training_survives_watchdog_cancel():
-    """A cancelled run() must stop the training thread, not orphan it (#5075 follow-up).
+def _make_run_component(predictor, now_utc):
+    """Build a training component that run() can drive to the point of launching a fine-tune.
+
+    The fetch, model status, prediction and publish steps are stubbed to record that they ran, so a
+    test can tell which of them a tick did.
+    """
+    component = _make_training_component(predictor, now_utc)
+    component.base.midnight_utc = now_utc
+    component.base.prediction_started = False
+    component.ml_enable = True
+    component.database_history_loaded = True
+    component.load_ml_database_days = 0
+    component.data_ready = True
+    component.ml_min_days = 1
+    component.initial_training_done = True
+    component.touched = []
+    component.logged = []
+    component.log = component.logged.append
+
+    async def fake_fetch(*args, **kwargs):
+        """Record a fetch."""
+        component.touched.append("fetch")
+
+    component._do_fetch = fake_fetch
+    component._update_model_status = lambda *args, **kwargs: component.touched.append("status")
+    component._get_predictions = lambda *args, **kwargs: component.touched.append("predict")
+    component._publish_entity = lambda *args, **kwargs: component.touched.append("publish")
+    return component
+
+
+async def _wait_for_event(event, timeout=10.0):
+    """Yield to the loop until a threading.Event is set or the timeout passes; return whether it was set."""
+    import asyncio
+    import time
+
+    deadline = time.monotonic() + timeout
+    while not event.is_set() and time.monotonic() < deadline:
+        await asyncio.sleep(0.01)
+    return event.is_set()
+
+
+def _test_component_training_runs_in_background():
+    """run() must launch training as a background task and keep predicting while it runs.
+
+    Awaiting the run would put it under run()'s run_timeout, and the watchdog would cancel - and so
+    throw away - any curriculum longer than that, which a first run on slow hardware can
+    legitimately be, so that model would never exist. Because the run trains a clone, ticks in the
+    meantime keep refreshing the forecast from the live model; they must just not start a second
+    run. Once it finishes, the next tick must publish from fresh data.
+    """
+    import asyncio
+    import threading
+    from load_ml_component import PREDICTION_INTERVAL_SECONDS
+
+    now_utc = datetime.now(timezone.utc)
+    entered = threading.Event()
+    release = threading.Event()
+    observed = {"calls": 0}
+
+    class BlockingPredictor(_SelfCloningFake):
+        """Stays 'in training' until the test releases it."""
+
+        def train_curriculum(self, *args, **kwargs):
+            """Block until released, then report a successful run."""
+            observed["calls"] += 1
+            entered.set()
+            observed["released"] = release.wait(timeout=10.0)
+            return 0.05
+
+    component = _make_run_component(BlockingPredictor(), now_utc)
+
+    async def scenario():
+        """Launch, tick while training, finish, then tick again."""
+        observed["launch_result"] = await component.run(PREDICTION_INTERVAL_SECONDS, first=False)
+        observed["launch_touched"] = list(component.touched)
+        observed["entered"] = await _wait_for_event(entered)
+        observed["status_while_training"] = component.training_status()
+
+        component.touched.clear()
+        observed["busy_result"] = await component.run(PREDICTION_INTERVAL_SECONDS, first=False)
+        observed["busy_touched"] = list(component.touched)
+
+        release.set()
+        await component.training_task
+
+        component.touched.clear()
+        observed["after_result"] = await component.run(60, first=False)
+        observed["after_touched"] = list(component.touched)
+
+    asyncio.run(scenario())
+
+    full_cycle = ["fetch", "status", "predict", "publish"]
+    assert observed["launch_result"] is True, "the tick that launches training must succeed"
+    assert observed["entered"], "the training run must start in the background after run() returns"
+    assert observed["launch_touched"] == full_cycle, f"the launching tick must still predict and publish, got {observed['launch_touched']}"
+    assert observed["status_while_training"] == "training", f"training_status must report the run in flight, got {observed['status_while_training']}"
+    assert observed["busy_result"] is True, "a tick during training must succeed"
+    assert observed["busy_touched"] == full_cycle, f"a 30-minute tick during training must keep refreshing the forecast from the live model, got {observed['busy_touched']}"
+    assert observed["calls"] == 1, f"a tick during training must not start a second run, got {observed['calls']} runs"
+    assert observed.get("released") is True, "the trainer must have been released by the test, not timed out"
+    assert component.last_train_time is not None, "the background run must record its completion"
+    assert observed["after_result"] is True, "the tick after training must succeed"
+    assert observed["after_touched"] == full_cycle, f"the tick after training must re-fetch and publish the new model even off the 30-minute boundary, got {observed['after_touched']}"
+    assert component.training_task is None and component.training_status() == "idle", "the finished task must be collected"
+
+    return True
+
+
+def _test_component_initial_training_is_published():
+    """Ticks during a first training run must keep the stats sensor current, with no model yet.
 
-    Now that training is off the loop, ComponentBase.start()'s run_timeout watchdog can actually
-    fire, and its task.cancel() reaches the coroutine awaiting to_thread - not the worker thread
-    already inside train_curriculum. api_stop is still False on that path, so without the
-    component's own cancel signal the thread would train on unwatched, and the next 60s tick would
-    start a second curriculum mutating the same predictor's weights and Adam state in place.
+    There is nothing to predict with until the first run lands, so these ticks skip the prediction
+    cycle - but a first run on slow hardware is exactly the long one whose progress, or stall, the
+    sensor is there to show, so they must still publish on the usual 30-minute cycle, and the
+    component must stay alive.
     """
     import asyncio
     import threading
+    from load_ml_component import PREDICTION_INTERVAL_SECONDS
+
+    now_utc = datetime.now(timezone.utc)
+    release = threading.Event()
+
+    class BlockingPredictor(_SelfCloningFake):
+        """Stays 'in training' until the test releases it."""
+
+        model_initialized = False
+
+        def train_curriculum(self, *args, **kwargs):
+            """Block until released."""
+            release.wait(timeout=10.0)
+            return None
+
+    component = _make_run_component(BlockingPredictor(), now_utc)
+    component.initial_training_done = False
+    observed = {}
+
+    async def scenario():
+        """Launch the first run, then tick on and off the 30-minute boundary."""
+        await component.run(PREDICTION_INTERVAL_SECONDS, first=False)
+        component.touched.clear()
+        component.last_success_timestamp = None
+        await component.run(PREDICTION_INTERVAL_SECONDS, first=False)
+        observed["boundary"] = list(component.touched)
+        observed["alive"] = component.last_success_timestamp is not None
+        component.touched.clear()
+        await component.run(60, first=False)
+        observed["off_boundary"] = list(component.touched)
+        release.set()
+        await component.training_task
+
+    asyncio.run(scenario())
+
+    assert observed["boundary"] == ["fetch", "status", "publish"], f"a 30-minute tick during the first run must publish its status without predicting, got {observed['boundary']}"
+    assert observed["alive"], "a tick during the first run must keep the component alive"
+    assert observed["off_boundary"] == [], f"ticks between 30-minute boundaries must not publish, got {observed['off_boundary']}"
+
+    return True
+
+
+def _test_component_training_stall_and_slow_run():
+    """An in-flight run is judged on progress: a slow one is logged once, a stalled one abandoned.
+
+    No run_timeout covers the background run, so this is the only thing that notices one stuck
+    inside the trainer. A stall must be logged as an error, ask the thread to abandon, and show on
+    the stats sensor until the run actually ends.
+    """
+    import asyncio
     import time
+    from load_ml_component import TRAINING_STALL_SECONDS, TRAINING_SLOW_WARN_SECONDS
+
+    now_utc = datetime.now(timezone.utc)
+    component = _make_run_component(_SelfCloningFake(), now_utc)
+    component.last_train_time = now_utc  # No fine-tune due, so the ticks below only watch
+    component.training_running = True  # The worker thread is inside the trainer
+    now_monotonic = time.monotonic()
+
+    # Slow but progressing: past the warning threshold, last epoch a few minutes ago
+    component.training_started_monotonic = now_monotonic - TRAINING_SLOW_WARN_SECONDS - 60
+    component.training_progress_monotonic = now_monotonic - 5 * 60
+    for _ in range(2):
+        assert asyncio.run(component.run(60, first=False)) is True, "a tick during a slow run is not an error"
+    slow_warnings = [msg for msg in component.logged if "still progressing" in msg]
+    assert len(slow_warnings) == 1, f"a slow run must be logged exactly once, got {slow_warnings}"
+    assert component.training_stalled is False and component.training_cancelled is False, "a run that is still progressing must be left to finish"
+
+    # Stalled: no epoch for longer than the stall window
+    component.training_progress_monotonic = now_monotonic - TRAINING_STALL_SECONDS - 60
+    component.touched.clear()
+    for _ in range(2):
+        assert asyncio.run(component.run(60, first=False)) is True, "a tick during a stalled run is not itself an error"
+    stall_errors = [msg for msg in component.logged if msg.startswith("Error:") and "no progress" in msg]
+    assert len(stall_errors) == 1, f"a stall must be logged as an error exactly once, got {stall_errors}"
+    assert component.training_stalled is True, "a run with no progress for the stall window must be flagged"
+    assert component.training_stop_requested() is True, "the trainer's stop hook must answer True once the run is stalled"
+    assert component.training_status() == "stalled", "the stats sensor must show the stall"
+    assert component.touched.count("publish") == 1, f"the stall must be published straight away, once, even off the 30-minute boundary, got {component.touched}"
+
+    # The stuck thread finally gets out: the tick that collects its task clears the flag
+    async def thread_exits():
+        """Finish the run's task and tick."""
+        component.training_running = False
+        component.training_task = asyncio.get_running_loop().create_future()
+        component.training_task.set_result(None)
+        return await component.run(60, first=False)
+
+    assert asyncio.run(thread_exits()) is True, "the tick that collects a stalled run must succeed"
+    assert component.training_stalled is False and component.training_status() == "idle", "once the stalled run has ended the sensor must stop reporting a stall"
+
+    return True
+
+
+def _test_component_training_retry_backoff():
+    """A failed run must be retried after TRAINING_RETRY_SECONDS, not relaunched on the next tick.
+
+    Now that a run's failure returns straight to a free loop, a run that fails every time - or one
+    abandoned for stalling - would otherwise be relaunched every minute.
+    """
+    import asyncio
+    from load_ml_component import TRAINING_RETRY_SECONDS
+
+    now_utc = datetime.now(timezone.utc)
+    calls = {"count": 0}
+
+    class FailingPredictor(_SelfCloningFake):
+        """Fails its first two runs without being asked to stop, then succeeds."""
+
+        def train_curriculum(self, *args, **kwargs):
+            """Report every pass failed, until the third run."""
+            calls["count"] += 1
+            return 0.05 if calls["count"] >= 3 else None
+
+    component = _make_run_component(FailingPredictor(), now_utc)
+
+    async def scenario():
+        """Fail once, tick inside the backoff, then tick after it."""
+        await component.run(60, first=False)
+        await component.training_task
+        await component.run(60, first=False)
+        calls["inside_backoff"] = calls["count"]
+        component.base.now_utc = now_utc + timedelta(seconds=TRAINING_RETRY_SECONDS + 60)
+        await component.run(60, first=False)
+        calls["retry_after_while_retrying"] = component.training_retry_after
+        await component.training_task
+        component.base.now_utc = now_utc + timedelta(seconds=2 * TRAINING_RETRY_SECONDS + 120)
+        await component.run(60, first=False)
+        await component.training_task
+
+    asyncio.run(scenario())
+
+    assert calls["inside_backoff"] == 1, f"a failed run must not be relaunched inside the backoff, got {calls['inside_backoff']} runs"
+    assert calls["count"] == 3, f"a failed run must be retried each time its backoff has passed, got {calls['count']} runs"
+    assert calls["retry_after_while_retrying"] is None, "launching the retry must clear the retry time, or the sensor shows a retry pending while it runs"
+    assert component.training_retry_after is None, "a successful run must leave no retry pending"
+    assert any("Will retry training" in msg for msg in component.logged), "the retry time must be logged"
+
+    return True
+
+
+def _test_component_training_survives_task_cancel():
+    """A cancelled training task must stop the training thread, not orphan it (#5075 follow-up).
+
+    Cancelling the task - the component's loop being torn down at shutdown, say - reaches the
+    coroutine awaiting to_thread, not the worker thread already inside train_curriculum. api_stop
+    need not be set on that path, so without the component's own cancel signal the thread would
+    train on unwatched. And until it finishes, run() must not start a second curriculum alongside
+    it: the two would compete for the CPU, and the new run would reset the cancel signal and tell
+    the orphan to carry on.
+    """
+    import asyncio
+    import threading
+    import time
+    from load_ml_component import PREDICTION_INTERVAL_SECONDS
 
     now_utc = datetime.now(timezone.utc)
     entered = threading.Event()
@@ -3759,10 +4038,10 @@ def train_curriculum(self, *args, **kwargs):
             allow_exit.wait(timeout=10.0)
             return None
 
-    component = _make_training_component(CancelWatchPredictor(), now_utc)
+    component = _make_run_component(CancelWatchPredictor(), now_utc)
 
     async def scenario():
-        """Cancel the training coroutine mid-run, then try to start another cycle."""
+        """Cancel the training coroutine mid-run, then tick with a retrain due."""
         task = asyncio.ensure_future(component._do_training(is_initial=False))
         for _ in range(200000):
             if entered.is_set():
@@ -3776,12 +4055,13 @@ async def scenario():
         except asyncio.CancelledError:
             observed["cancelled"] = True
 
-        # The worker thread is still inside train_curriculum here. run()'s finally has already
-        # cleared load_ml_calculating, so is_calculating() has to get its answer from elsewhere,
-        # and the next run() tick must not start a second training on the same predictor.
+        # The worker thread is still inside train_curriculum here, so is_calculating() has to get
+        # its answer from it, and a tick with a retrain due (last_train_time is None) must not
+        # start a second training
         component.load_ml_calculating = False
         observed["calculating_while_orphaned"] = component.is_calculating()
-        await component._do_training(is_initial=False)
+        observed["orphan_tick"] = await component.run(PREDICTION_INTERVAL_SECONDS, first=False)
+        observed["orphan_tick_task"] = component.training_task
         allow_exit.set()
 
         for _ in range(1000):
@@ -3791,10 +4071,11 @@ async def scenario():
 
     asyncio.run(scenario())
 
-    assert observed.get("cancelled") is True, "cancelling the run() task must propagate out of _do_training rather than being swallowed"
+    assert observed.get("cancelled") is True, "cancelling the training task must propagate out of _do_training rather than being swallowed"
     assert component.training_cancelled is True, "a cancelled run must raise the component's own cancel signal, or the worker thread never learns to abandon"
     assert observed.get("hook_tripped") is True, "the stop hook must report True after the coroutine is cancelled, so the orphaned training thread abandons its run"
     assert observed["calls"] == 1, f"a second training must not start while the previous worker thread is still running, got {observed['calls']} calls"
+    assert observed.get("orphan_tick") is True and observed.get("orphan_tick_task") is None, "a tick during an orphaned run must succeed without launching a new training task"
     assert observed.get("calculating_while_orphaned") is True, "is_calculating() must still report True while an orphaned training thread is doing NumPy work"
     assert component.training_running is False, "the worker thread must clear training_running when it returns, or training never runs again"
     assert component.last_train_time is None, "a cancelled run must not stamp last_train_time"
diff --git a/docs/components.md b/docs/components.md
index 452448dcf..18cc8444e 100644
--- a/docs/components.md
+++ b/docs/components.md
@@ -1737,6 +1737,11 @@ The ML component tracks several status indicators:
 - **validation_failed**: Predictions are disabled due to high validation error
 - **stale**: Model hasn't been trained in 48+ hours and needs retraining
 
+Training runs in the background on a copy of the model, so the current model keeps producing predictions every 30 minutes while it does, and is replaced only once a run succeeds.
+The `training_status` attribute of `sensor.predbat_load_ml_stats` shows `idle`, `training` or `stalled`, with `training_started` and `training_last_progress` timestamps as of the last prediction cycle, and `training_retry_after` once a run has failed.
+A first training run on slow hardware can take hours: Predbat logs a warning once a run has gone two hours and is still progressing, and lets it finish.
+A run that completes no training epoch for two hours is treated as stalled: the error is logged, the run is abandoned and retried 30 minutes after it ends, and the current model stays in use.
+
 Check Predbat logs for training progress and validation metrics:
 
 ```text