Merge pull request #4 from FizzWizZleDazzle/weights-hot-reload

Hot-reload weights from the database; deploys self-promote

Author: FizzWizZleDazzle

docs/design.md

@@ -224,6 +224,12 @@ in both splits, and held-out AUC within 0.005 of the incumbent. Promotion
 inserts a new row in `weights_versions` and flips `active`; rollback flips
 it back. Weights never mutate live.
 
+The active row is the record and reaches running pods without a restart:
+the service polls it once a minute and swaps in a changed table. On
+startup the embedded table self-promotes when its `fitted_at` is newer
+than the active row's, so a deploy carrying a retune needs no manual
+promotion; an older or equal embedded fit defers to the database.
+
 ## Federation (`federation.rs`, dormant)
 
 Two record kinds with different physics:

src/engine.rs

@@ -81,7 +81,7 @@ impl Thresholds {
 
 /// The weight table: bias, per-rule weights, tier thresholds. Serialized as
 /// JSON and shipped as data, not code.
-#[derive(Debug, Clone, Serialize, Deserialize)]
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
 pub struct Weights {
     pub bias: f64,
     pub rules: BTreeMap<String, f64>,
@@ -153,6 +153,28 @@ impl Weights {
         serde_json::from_str(include_str!("../weights/default.json"))
             .expect("embedded default weights must parse")
     }
+
+    /// Fit timestamp from the provenance meta; None when the table
+    /// carries none.
+    pub fn fitted_at(&self) -> Option<chrono::DateTime<chrono::Utc>> {
+        self.meta
+            .as_ref()?
+            .get("fitted_at")?
+            .as_str()
+            .and_then(|s| chrono::DateTime::parse_from_rfc3339(s).ok())
+            .map(|t| t.with_timezone(&chrono::Utc))
+    }
+
+    /// Whether this table is a strictly newer fit than `other`. A table
+    /// without provenance is never newer, and any fitted table is newer
+    /// than one without provenance.
+    pub fn newer_fit_than(&self, other: &Weights) -> bool {
+        match (self.fitted_at(), other.fitted_at()) {
+            (Some(a), Some(b)) => a > b,
+            (Some(_), None) => true,
+            (None, _) => false,
+        }
+    }
 }
 
 #[cfg(test)]
@@ -276,6 +298,24 @@ mod tests {
         assert!(t.bias < 0.0, "prior must favor pass");
     }
 
+    #[test]
+    fn newer_fit_wins_and_missing_provenance_never_does() {
+        let stamped = |at: &str| Weights {
+            meta: Some(serde_json::json!({ "fitted_at": at })),
+            ..table()
+        };
+        let old = stamped("2026-08-06T21:24:28Z");
+        let new = stamped("2026-08-10T00:00:25Z");
+        assert!(new.newer_fit_than(&old));
+        assert!(!old.newer_fit_than(&new));
+        assert!(!old.newer_fit_than(&old));
+        let bare = table();
+        assert!(old.newer_fit_than(&bare));
+        assert!(!bare.newer_fit_than(&old));
+        assert!(!bare.newer_fit_than(&bare));
+        assert!(Weights::default_table().fitted_at().is_some());
+    }
+
     #[test]
     fn weights_serde_roundtrip() {
         let t = table();

src/main.rs

@@ -47,18 +47,28 @@ async fn main() -> anyhow::Result<()> {
     let database_url = env("DATABASE_URL")?;
     let store = PgStore::connect(&database_url).await?;
 
+    // The active row in the database is the record; the embedded table
+    // self-promotes over it when it is a newer fit, so a deploy carrying
+    // retuned weights needs no manual promotion step.
+    let embedded = Weights::default_table();
     let weights = match store.active_weights().await? {
-        Some(w) => {
+        Some(active) if embedded.newer_fit_than(&active) => {
+            store
+                .promote_weights(&embedded, "newer embedded fit", f64::NAN)
+                .await?;
+            info!("promoted newer embedded weights over the active row");
+            embedded
+        }
+        Some(active) => {
             info!("loaded active weights from database");
-            w
+            active
         }
         None => {
-            let w = Weights::default_table();
             store
-                .promote_weights(&w, "bootstrap defaults", f64::NAN)
+                .promote_weights(&embedded, "bootstrap defaults", f64::NAN)
                 .await?;
             info!("promoted embedded default weights");
-            w
+            embedded
         }
     };
 
@@ -96,6 +106,30 @@ async fn main() -> anyhow::Result<()> {
         });
     }
 
+    // Weight hot-reload: promotions land in the database (nightly
+    // learner, operator, another replica); running pods pick them up
+    // without a restart.
+    {
+        let state = state.clone();
+        tokio::spawn(async move {
+            let mut tick = tokio::time::interval(std::time::Duration::from_secs(60));
+            tick.tick().await;
+            loop {
+                tick.tick().await;
+                match state.store.active_weights().await {
+                    Ok(Some(w)) => {
+                        if w != *state.weights.read().await {
+                            info!("active weights changed; hot-reloading");
+                            *state.weights.write().await = w;
+                        }
+                    }
+                    Ok(None) => {}
+                    Err(e) => warn!("weights reload: {e:#}"),
+                }
+            }
+        });
+    }
+
     let app = Router::new()
         .route("/healthz", get(|| async { "ok" }))
         .route("/webhook", post(handle_webhook))