feat(operator): publish and route discovered model inventories

.github/workflows/helm.yaml

@@ -754,6 +754,11 @@ jobs:
           - peer-trust-mode: pin
             signed-state: enforce
             leg: pin, signed state
+          - peer-trust-mode: pin
+            hub-router: operator
+            signed-state: enforce
+            model-gossip: true
+            leg: pin, signed model gossip routing
     timeout-minutes: 60
     permissions:
       contents: read
@@ -817,6 +822,7 @@ jobs:
           HUB_ROUTER: ${{ matrix.hub-router || 'render' }}
           HA: ${{ matrix.ha || '0' }}
           SIGNED_STATE: ${{ matrix.signed-state || 'off' }}
+          MODEL_GOSSIP_TEST: ${{ matrix.model-gossip && '1' || '0' }}
           GRID_FEEDBACK_CONTRACT_TEST: ${{ matrix.peer-trust-mode == 'pin' && '1' || '0' }}
           # The spiffe leg rotates 8 minute identities; pin trust does not rotate.
           ROTATION_LIFETIME: "480"

charts/grid-enrollment/tests/cert-manager-e2e.sh

@@ -41,6 +41,27 @@ kubectl apply --server-side -f "$WORK/cert-manager.yaml"
 for deployment in cert-manager cert-manager-cainjector cert-manager-webhook; do
   kubectl -n cert-manager rollout status "deployment/$deployment" --timeout=180s
 done
+webhook_deadline=$((SECONDS + 60))
+webhook_ready=false
+while ((SECONDS < webhook_deadline)); do
+  if kubectl create --dry-run=server -n cert-manager -f - >/dev/null 2>&1 <<'YAML'
+apiVersion: cert-manager.io/v1
+kind: Issuer
+metadata:
+  name: webhook-readiness-check
+spec:
+  selfSigned: {}
+YAML
+  then
+    webhook_ready=true
+    break
+  fi
+  sleep 1
+done
+[[ "$webhook_ready" == true ]] || {
+  echo "cert-manager webhook did not accept a server-side Issuer dry run" >&2
+  exit 1
+}
 kubectl create namespace fresh
 kubectl create namespace existing
 

crates/gateway/Cargo.toml

@@ -25,6 +25,7 @@ httpdate = "1.0.3"
 serde = { version = "1.0.228", features = ["derive"] }
 serde_yaml = { version = "0.10.7", package = "yaml_serde" }
 metrics = "0.24.6"
+time = "0.3.0"
 
 [workspace.lints.rust]
 dead_code = "deny"

crates/gateway/crates/ai-grid-filters/Cargo.toml

@@ -36,6 +36,7 @@ xxhash-rust = { version = "0.8.0", features = ["xxh64"] }
 lru = "0.18.0"
 foldhash = "0.2.0"
 bytes = "1.0.0"
+time = { workspace = true, features = ["parsing"] }
 tokio = { version = "1.53.1", features = ["net", "rt", "time"] }
 
 [dev-dependencies]

crates/gateway/crates/ai-grid-filters/src/descriptor.rs

@@ -445,6 +445,8 @@ fn validate_header_value(field: &str, value: &str) -> Result<(), FilterError> {
     reason = "tests"
 )]
 mod tests {
+    use time::{OffsetDateTime, format_description::well_known::Rfc3339};
+
     use super::*;
 
     fn candidate(kind_str: &str, name: &str, site: &str, cluster: &str) -> CandidateConfig {
@@ -463,6 +465,30 @@ mod tests {
         }
     }
 
+    #[test]
+    fn inventory_expiry_is_enforced_at_the_clock_lag_boundary() {
+        let deadline = OffsetDateTime::parse("2030-01-01T00:01:00Z", &Rfc3339).unwrap();
+        let deadline_ms = u64::try_from(deadline.unix_timestamp_nanos().div_euclid(1_000_000)).unwrap();
+        let deadline_ms_i64 = i64::try_from(deadline_ms).unwrap();
+        let margin_ms = i64::try_from(grid_signals::CANDIDATE_EXPIRY_SKEW_MARGIN_MS).unwrap();
+        let mut raw = candidate("inference_model", "m", "site-a", "pool-a");
+        raw.expires_at_ms = Some(deadline_ms);
+        let candidates = validate_serving_candidates(vec![raw]).unwrap();
+        let candidate = candidates.first().unwrap();
+        assert!(
+            !candidate.expired_at(deadline_ms_i64.saturating_sub(margin_ms).saturating_sub(1)),
+            "route remains usable before the safety margin"
+        );
+        assert!(
+            candidate.expired_at(deadline_ms_i64.saturating_sub(margin_ms)),
+            "route expires exactly at the safety margin"
+        );
+        assert!(
+            candidate.expired_at(deadline_ms_i64),
+            "deadline never receives a grace period"
+        );
+    }
+
     #[test]
     fn valid_minimal_inference_candidate() {
         let candidates =

crates/gateway/crates/ai-grid-filters/src/route.rs

@@ -96,12 +96,37 @@ enum RouteDecision {
     Select,
 }
 
-/// Resolve the empty revision before preserving a prior filter's selection.
-fn route_decision(snapshot: &RouteSnapshot, earlier_selected: bool) -> RouteDecision {
+/// Resolve withdrawals before preserving a prior filter's selection; scope inventory revisions to the request model.
+fn route_decision(snapshot: &RouteSnapshot, earlier_selected: bool, model: Option<&str>, now_ms: i64) -> RouteDecision {
     if snapshot.candidates.is_empty() {
         RouteDecision::NoRoute
     } else if earlier_selected {
-        RouteDecision::KeepEarlier
+        let Some(model) = model else {
+            return RouteDecision::KeepEarlier;
+        };
+        let matching =
+            |candidate: &RouteCandidate| candidate.kind == CapabilityKind::InferenceModel && &*candidate.name == model;
+        if !snapshot.candidates.iter().any(&matching) {
+            RouteDecision::KeepEarlier
+        } else if snapshot
+            .candidates
+            .iter()
+            .filter(|candidate| matching(candidate))
+            .all(|candidate| candidate.expired_at(now_ms))
+        {
+            RouteDecision::NoRoute
+        } else if snapshot
+            .candidates
+            .iter()
+            .filter(|candidate| matching(candidate))
+            .all(|candidate| candidate.expires_at_ms.is_none())
+        {
+            RouteDecision::KeepEarlier
+        } else {
+            RouteDecision::Select
+        }
+    } else if snapshot.candidates.iter().all(|candidate| candidate.expired_at(now_ms)) {
+        RouteDecision::NoRoute
     } else {
         RouteDecision::Select
     }
@@ -754,19 +779,23 @@ impl HttpFilter for GridSiteRouteFilter {
                 .push(http::header::HeaderName::from_static(name));
         }
         let snapshot = self.snapshot.load();
-        match route_decision(&snapshot, ctx.cluster.is_some() || ctx.upstream.is_some()) {
+        let earlier_selected = ctx.cluster.is_some() || ctx.upstream.is_some();
+        let model = ctx
+            .request
+            .headers
+            .get(&self.model_header)
+            .and_then(|value| value.to_str().ok());
+        match route_decision(&snapshot, earlier_selected, model, grid_signals::now_ms()) {
             RouteDecision::NoRoute => {
                 return Ok(refuse(Refused::NoRoute, "", self.turn.fetch_add(1, Ordering::Relaxed)));
             },
             RouteDecision::KeepEarlier => return Ok(FilterAction::Continue),
-            RouteDecision::Select => {},
+            RouteDecision::Select if earlier_selected && model.is_none() => return Ok(FilterAction::Continue),
+            RouteDecision::Select => {
+                ctx.cluster = None;
+                ctx.upstream = None;
+            },
         }
-        // A copy of the request reference, so the model outlives writes to ctx.
-        let request = ctx.request;
-        let model = request
-            .headers
-            .get(&self.model_header)
-            .and_then(|value| value.to_str().ok());
         if let Some(action) = self.pin(ctx, model)? {
             return Ok(action);
         }
@@ -993,7 +1022,7 @@ fn pinned_target<'snap>(
         return Target::Gone;
     }
     at_site
-        .find(|candidate| is_admitted_for_new_request(candidate.admission_state))
+        .find(|candidate| is_admitted_for_new_request(candidate.admission_state) && !candidate.expired_at(now_ms))
         .map_or(Target::Unavailable, Target::Found)
 }
 
@@ -1312,6 +1341,7 @@ fn matching<'snap, 'req>(
 where
     'snap: 'req,
 {
+    let now_ms = grid_signals::now_ms();
     snapshot
         .candidates
         .iter()
@@ -1322,6 +1352,7 @@ where
             candidate.kind == kind
                 && &*candidate.name == name
                 && is_admitted_for_new_request(candidate.admission_state)
+                && !candidate.expired_at(now_ms)
                 && routable(candidate)
         })
 }
@@ -1534,27 +1565,10 @@ pub(crate) mod tests {
     fn queued(sites: &[(&str, f64, AdmissionState)]) -> RouteSnapshot {
         let mut sorted = sites.to_vec();
         sorted.sort_by(|left, right| left.1.total_cmp(&right.1));
-        let mut candidates = validate_candidates(
-            sorted
-                .iter()
-                .map(|(site, ..)| CandidateConfig {
-                    expires_at_ms: None,
-                    stable_id: None,
-                    cluster: format!("pool-{site}"),
-                    credential: None,
-                    admission: AdmissionState::default(),
-                    fresh: true,
-                    kind: CapabilityKind::InferenceModel,
-                    name: "llama".to_owned(),
-                    publishes_load: true,
-                    site: (*site).to_owned(),
-                })
-                .collect(),
-        )
-        .unwrap();
-        for (candidate, (_, _, admission)) in candidates.iter_mut().zip(&sorted) {
-            candidate.admission_state = *admission;
-        }
+        let candidates = sorted
+            .iter()
+            .flat_map(|(site, _, admission)| one("llama", site, &format!("pool-{site}"), *admission))
+            .collect();
         let mut snapshot = RouteSnapshot::from_static(candidates, Arc::from("a"));
         snapshot.scores = sorted.iter().map(|(_, queue, _)| *queue).collect();
         snapshot
@@ -1671,12 +1685,13 @@ pub(crate) mod tests {
     #[test]
     fn discovered_candidates_stop_routing_before_expiry_with_clock_skew_margin() {
         let mut candidates = one("llama", "a", "pool-a", AdmissionState::NewAndExisting);
+        let now_ms = grid_signals::now_ms();
         if let Some(candidate) = candidates.first_mut() {
-            candidate.expires_at_ms = Some(100_000);
+            candidate.expires_at_ms = Some(u64::try_from(now_ms.saturating_add(300_000)).unwrap_or(u64::MAX));
         }
         let snapshot = RouteSnapshot::from_static(candidates, Arc::from("a"));
-        let routable_before = |candidate: &RouteCandidate| !candidate.expired_at(69_999);
-        let routable_after = |candidate: &RouteCandidate| !candidate.expired_at(70_000);
+        let routable_before = |candidate: &RouteCandidate| !candidate.expired_at(now_ms.saturating_add(269_999));
+        let routable_after = |candidate: &RouteCandidate| !candidate.expired_at(now_ms.saturating_add(270_000));
 
         assert!(
             select_spread(
@@ -1703,7 +1718,12 @@ pub(crate) mod tests {
             "the route is withdrawn at the deadline minus the 30-second skew margin"
         );
         assert!(matches!(
-            pinned_target(&snapshot, Some(&holder("a", Some("pool-a"))), Some("llama"), 70_000),
+            pinned_target(
+                &snapshot,
+                Some(&holder("a", Some("pool-a"))),
+                Some("llama"),
+                now_ms.saturating_add(270_000)
+            ),
             Target::Gone
         ));
     }
@@ -1982,6 +2002,59 @@ pub(crate) mod tests {
         assert_eq!(unserved(&excluded, "llama", |_| false), Refused::NotReady);
     }
 
+    #[test]
+    fn expired_inventory_cannot_route_from_a_retained_snapshot() {
+        let mut candidates = one("llama", "east", "expired-pool", AdmissionState::NewAndExisting);
+        let now_ms = grid_signals::now_ms();
+        candidates[0].expires_at_ms = Some(u64::try_from(now_ms.saturating_add(29_000)).unwrap_or(u64::MAX));
+        let snapshot = RouteSnapshot::from_static(candidates.clone(), Arc::from("east"));
+        assert!(
+            select_spread(
+                &snapshot,
+                CapabilityKind::InferenceModel,
+                "llama",
+                0,
+                |_| true,
+                &KeepAll
+            )
+            .is_none(),
+            "expired inventory cannot be selected even without a config reload"
+        );
+        assert!(
+            matches!(pinned_target(&snapshot, None, Some("llama"), now_ms), Target::Gone),
+            "stored-state pins cannot bypass inventory expiry"
+        );
+        assert_eq!(
+            route_decision(&snapshot, true, Some("llama"), now_ms),
+            RouteDecision::NoRoute,
+            "an earlier filter choice cannot bypass an expired snapshot"
+        );
+        candidates.extend(one("llama", "west", "static-pool", AdmissionState::NewAndExisting));
+        assert_eq!(
+            front(candidates, CapabilityKind::InferenceModel, "llama").as_deref(),
+            Some("static-pool"),
+            "a static alternative remains routable after discovered inventory expires"
+        );
+    }
+
+    #[test]
+    fn fresh_inventory_routes_and_revalidates_an_earlier_selection() {
+        let mut candidates = one("llama", "east", "discovered-pool", AdmissionState::NewAndExisting);
+        let now_ms = grid_signals::now_ms();
+        candidates[0].expires_at_ms = Some(u64::try_from(now_ms.saturating_add(600_000)).unwrap_or(u64::MAX));
+        let snapshot = RouteSnapshot::from_static(candidates.clone(), Arc::from("east"));
+        assert_eq!(
+            route_decision(&snapshot, true, Some("llama"), now_ms),
+            RouteDecision::Select,
+            "dynamic routes require deadline-aware selection even after another filter chose a cluster"
+        );
+        assert_eq!(
+            front(candidates, CapabilityKind::InferenceModel, "llama").as_deref(),
+            Some("discovered-pool"),
+            "unexpired inventory remains routable"
+        );
+    }
+
     #[test]
     fn front_match_is_selected() {
         let candidates = one("llama", "east", "pool-a", AdmissionState::NewAndExisting);
@@ -2594,13 +2667,19 @@ pub(crate) mod tests {
     #[test]
     fn empty_snapshot_rejects_even_with_a_preselected_cluster() {
         let empty = RouteSnapshot::from_static(Vec::new(), Arc::from("local"));
-        assert!(matches!(route_decision(&empty, true), RouteDecision::NoRoute));
+        assert!(matches!(
+            route_decision(&empty, true, Some("llama"), TEST_NOW_MS),
+            RouteDecision::NoRoute
+        ));
 
         let active = RouteSnapshot::from_static(
             one("llama", "east", "pool-a", AdmissionState::NewAndExisting),
             Arc::from("local"),
         );
-        assert!(matches!(route_decision(&active, true), RouteDecision::KeepEarlier));
+        assert!(matches!(
+            route_decision(&active, true, Some("llama"), TEST_NOW_MS),
+            RouteDecision::KeepEarlier
+        ));
     }
 
     #[test]
@@ -2728,6 +2807,42 @@ pub(crate) mod tests {
         filter
     }
 
+    #[tokio::test]
+    async fn a_request_without_a_model_keeps_an_earlier_route() {
+        let mut candidates = one("llama", "east", "pool-a", OPEN);
+        candidates[0].expires_at_ms = Some(u64::try_from(grid_signals::now_ms().saturating_add(60_000)).unwrap());
+        let filter = filter(
+            RouteSnapshot::from_static(candidates, Arc::from("local")),
+            AffinitySettings::default(),
+        );
+        let request = praxis_filter::Request {
+            headers: http::HeaderMap::new(),
+            method: http::Method::POST,
+            uri: http::Uri::from_static("/v1/chat/completions"),
+        };
+        let mut ctx = context(&request);
+        let earlier_dial = dial_to("earlier-cluster", "10.0.0.8:8443", 1);
+        ctx.cluster = Some(Arc::from("earlier-cluster"));
+        ctx.upstream = Some(earlier_dial.upstream().clone());
+
+        let action = filter.on_request(&mut ctx).await.expect("keeps earlier route");
+
+        assert!(
+            matches!(action, FilterAction::Continue),
+            "the prior filter's route continues"
+        );
+        assert_eq!(
+            ctx.cluster.as_deref(),
+            Some("earlier-cluster"),
+            "the earlier cluster remains selected"
+        );
+        assert_eq!(
+            ctx.upstream.as_ref().map(|upstream| &*upstream.address),
+            Some("10.0.0.8:8443"),
+            "the earlier upstream remains selected"
+        );
+    }
+
     /// One request for `llama` at `path` carrying a caller credential.
     fn llama_request(path: &'static str) -> praxis_filter::Request {
         praxis_filter::Request {