feat: continuous coverage evaluation + calibration-triggered retraining

Add a background loop that re-evaluates p90 coverage on the current model
and test buffer every COVERAGE_EVAL_INTERVAL_SEC, without retraining. When
the EMA of |coverage - target| crosses CALIBRATION_TRIGGER_THRESHOLD for K
consecutive evaluations, signal the training loop to retrain immediately
rather than waiting out the full RETRAINING_INTERVAL_SEC. Disabled by
default (COVERAGE_EVAL_INTERVAL_SEC=0).

Implements llm-d#19.

Signed-off-by: Greg Neighbors <26003+gkneighb@users.noreply.github.com>

Author: gkneighb

src/llm_d_latency_predictor/training_server.py

@@ -122,6 +122,17 @@ class Settings:
     TTFT_GATED_MODEL_PATH: str = os.getenv("LATENCY_TTFT_GATED_MODEL_PATH", "/tmp/models/ttft_gated.joblib")
     TPOT_GATED_MODEL_PATH: str = os.getenv("LATENCY_TPOT_GATED_MODEL_PATH", "/tmp/models/tpot_gated.joblib")
 
+    # Experiment C — continuous coverage evaluation + calibration-triggered retraining.
+    # When > 0, a background loop re-evaluates coverage on the current model + test
+    # buffer every COVERAGE_EVAL_INTERVAL_SEC. If the EMA of |coverage - quantile_alpha|
+    # exceeds CALIBRATION_TRIGGER_THRESHOLD for CALIBRATION_TRIGGER_K consecutive
+    # evaluations, the next retrain fires immediately instead of waiting out the
+    # full RETRAINING_INTERVAL_SEC. Set COVERAGE_EVAL_INTERVAL_SEC=0 to disable.
+    COVERAGE_EVAL_INTERVAL_SEC: int = int(os.getenv("LATENCY_COVERAGE_EVAL_INTERVAL_SEC", 0))
+    CALIBRATION_TRIGGER_THRESHOLD: float = float(os.getenv("LATENCY_CALIBRATION_TRIGGER_THRESHOLD", "5.0"))
+    CALIBRATION_TRIGGER_K: int = int(os.getenv("LATENCY_CALIBRATION_TRIGGER_K", 2))
+    CALIBRATION_EMA_ALPHA: float = float(os.getenv("LATENCY_CALIBRATION_EMA_ALPHA", "0.3"))
+
 
 class QueueGatedModel:
     """Wraps noqueue + queued sub-models into one joblib-serializable object.
@@ -320,6 +331,13 @@ def __init__(self, model_type: str = None):
         self.last_retrain_time = None
         self._shutdown_event = threading.Event()
         self._training_thread: threading.Thread = None
+        # Experiment C — calibration trigger. The continuous coverage loop sets
+        # this when |coverage - quantile_alpha| EMA crosses
+        # CALIBRATION_TRIGGER_THRESHOLD; the training loop checks it during its
+        # inter-retrain sleep and fires an immediate retrain rather than waiting
+        # the rest of RETRAINING_INTERVAL_SEC.
+        self._calibration_trigger = threading.Event()
+        self._coverage_eval_thread: threading.Thread | None = None
 
     def _get_prefix_bucket(self, prefix_score: float) -> int:
         """Map prefix cache score to bucket index."""
@@ -714,6 +732,45 @@ def _train_model_with_scaling(
             logging.error(f"Error in _train_model_with_scaling: {e}", exc_info=True)
             raise
 
+    def evaluate_current_coverage(self) -> tuple[float | None, float | None]:
+        """Re-evaluate coverage on the *currently loaded* model + test buffer,
+        without retraining. Appends to {ttft,tpot}_coverage_scores so /metrics
+        reflects up-to-date calibration between scheduled retrains. Returns
+        (ttft_cov, tpot_cov); either may be None if model/test data isn't ready.
+        Quantile-objective only — no-op under mean-objective.
+        """
+        if self.objective_type == ObjectiveType.MEAN:
+            return None, None
+        if not self.is_ready:
+            return None, None
+
+        with self.lock:
+            ttft_test = list(self.ttft_test_data)
+            tpot_test = list(self.tpot_test_data)
+            ttft_model = self.ttft_model
+            tpot_model = self.tpot_model
+            ttft_scaler = self.ttft_scaler
+            tpot_scaler = self.tpot_scaler
+
+        ttft_cov: float | None = None
+        tpot_cov: float | None = None
+
+        if ttft_test and ttft_model is not None:
+            _, cov, _ = self._calculate_metrics_on_test(ttft_model, ttft_scaler, ttft_test, "ttft", "actual_ttft_ms")
+            if cov is not None:
+                with self.lock:
+                    self.ttft_coverage_scores.append(cov)
+                ttft_cov = cov
+
+        if tpot_test and tpot_model is not None:
+            _, cov, _ = self._calculate_metrics_on_test(tpot_model, tpot_scaler, tpot_test, "tpot", "actual_tpot_ms")
+            if cov is not None:
+                with self.lock:
+                    self.tpot_coverage_scores.append(cov)
+                tpot_cov = cov
+
+        return ttft_cov, tpot_cov
+
     def _calculate_metrics_on_test(self, model, scaler, test_data, model_name, target_col):
         """Calculate metrics on test data.
 
@@ -1824,11 +1881,85 @@ def continuous_training_loop():
             predictor.train()
         except Exception:
             logging.error("Error in periodic retraining", exc_info=True)
-        if predictor._shutdown_event.wait(timeout=settings.RETRAINING_INTERVAL_SEC):
-            break
+
+        # Sleep the retrain interval in 1-second slices so a calibration trigger
+        # (set by continuous_coverage_loop) can short-circuit the wait. Shutdown
+        # still takes precedence.
+        slept = 0.0
+        slice_sec = 1.0
+        while slept < settings.RETRAINING_INTERVAL_SEC:
+            if predictor._shutdown_event.is_set():
+                logging.info("Training loop exiting (shutdown).")
+                return
+            if predictor._calibration_trigger.is_set():
+                predictor._calibration_trigger.clear()
+                logging.info(
+                    f"Retraining triggered by calibration deviation after {slept:.1f}s of "
+                    f"{settings.RETRAINING_INTERVAL_SEC}s scheduled interval"
+                )
+                break
+            time.sleep(slice_sec)
+            slept += slice_sec
     logging.info("Training loop exiting.")
 
 
+# --- Continuous Coverage Evaluation Loop (Experiment C) ---
+def continuous_coverage_loop():
+    """Periodically re-evaluates calibration of the currently loaded model
+    against the test buffer. Detects drift between scheduled retrains and
+    sets _calibration_trigger when an EMA of |coverage - quantile_alpha|
+    crosses CALIBRATION_TRIGGER_THRESHOLD for K consecutive evaluations.
+    No-op when COVERAGE_EVAL_INTERVAL_SEC <= 0.
+    """
+    if settings.COVERAGE_EVAL_INTERVAL_SEC <= 0:
+        logging.info("Continuous coverage evaluation disabled (COVERAGE_EVAL_INTERVAL_SEC=0).")
+        return
+
+    # Wait long enough for the first train() to populate models + initial coverage.
+    time.sleep(15)
+    target_pct = settings.QUANTILE_ALPHA * 100
+    alpha = settings.CALIBRATION_EMA_ALPHA
+    ttft_dev_ema = 0.0
+    tpot_dev_ema = 0.0
+    consecutive_bad = 0
+
+    logging.info(
+        f"Continuous coverage loop started "
+        f"(eval_interval={settings.COVERAGE_EVAL_INTERVAL_SEC}s, "
+        f"target={target_pct:.0f}%, threshold={settings.CALIBRATION_TRIGGER_THRESHOLD:.2f}pp, "
+        f"k={settings.CALIBRATION_TRIGGER_K}, ema_alpha={alpha:.2f})"
+    )
+
+    while not predictor._shutdown_event.is_set():
+        try:
+            ttft_cov, tpot_cov = predictor.evaluate_current_coverage()
+            if ttft_cov is not None:
+                ttft_dev_ema = (1 - alpha) * ttft_dev_ema + alpha * abs(ttft_cov - target_pct)
+            if tpot_cov is not None:
+                tpot_dev_ema = (1 - alpha) * tpot_dev_ema + alpha * abs(tpot_cov - target_pct)
+            max_dev = max(ttft_dev_ema, tpot_dev_ema)
+            if max_dev > settings.CALIBRATION_TRIGGER_THRESHOLD:
+                consecutive_bad += 1
+                logging.info(
+                    f"Coverage drift: ttft_cov={ttft_cov} tpot_cov={tpot_cov} "
+                    f"ttft_dev_ema={ttft_dev_ema:.2f} tpot_dev_ema={tpot_dev_ema:.2f} "
+                    f"consecutive_bad={consecutive_bad}/{settings.CALIBRATION_TRIGGER_K}"
+                )
+                if consecutive_bad >= settings.CALIBRATION_TRIGGER_K:
+                    logging.warning(
+                        f"Calibration trigger fired (max_dev={max_dev:.2f}pp > "
+                        f"threshold={settings.CALIBRATION_TRIGGER_THRESHOLD:.2f}pp). Requesting immediate retrain."
+                    )
+                    predictor._calibration_trigger.set()
+                    consecutive_bad = 0  # one trigger per drift event
+            else:
+                consecutive_bad = 0
+        except Exception:
+            logging.error("Error in continuous coverage loop", exc_info=True)
+        predictor._shutdown_event.wait(timeout=settings.COVERAGE_EVAL_INTERVAL_SEC)
+    logging.info("Continuous coverage loop exiting.")
+
+
 # --- FastAPI Events ---
 @app.on_event("startup")
 async def startup_event():
@@ -1839,6 +1970,13 @@ async def startup_event():
     t.start()
     logging.info("Background training started.")
 
+    # Experiment C — start continuous coverage evaluation only when enabled.
+    if settings.COVERAGE_EVAL_INTERVAL_SEC > 0:
+        ct = threading.Thread(target=continuous_coverage_loop, daemon=True)
+        predictor._coverage_eval_thread = ct
+        ct.start()
+        logging.info("Continuous coverage evaluation thread started.")
+
 
 @app.on_event("shutdown")
 async def shutdown_event():