[Router] Fleet-wide sampling contract 2/3: enforce and inject per request (#39001)

Co-authored-by: Kangyan Zhou <kangyan.zhou@radixark.ai>
Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>

experimental/sgl-router/src/server/error.rs

@@ -17,6 +17,21 @@ pub enum ApiError {
     #[error("model not found: {0}")]
     ModelNotFound(String),
 
+    /// A request refused by the fleet-wide sampling contract
+    /// (`--override-sampling-params` under `--sampling-param-conflict
+    /// reject`).
+    ///
+    /// Distinct from [`Self::BadRequest`] on purpose: rolling a contract out
+    /// across a fleet turns previously-served client traffic into 400s, and
+    /// the operator's first question is how much and on which parameter.
+    /// Folded into `bad_request` that is unanswerable — the code would be the
+    /// same one malformed JSON and a missing `model` field already emit.
+    /// `param` is a `&'static str` from
+    /// [`crate::config::SamplingField::wire_name`], which keeps it usable as a
+    /// bounded metric label.
+    #[error("{param} violates this deployment's sampling contract: {detail}")]
+    SamplingContract { param: &'static str, detail: String },
+
     /// Could not reach the upstream worker (connect refused, DNS, TLS, request
     /// build error). `source` captures the full anyhow chain for server-side
     /// logging; clients see a generic message.
@@ -114,6 +129,9 @@ impl ApiError {
     fn status_and_code(&self) -> (StatusCode, &'static str) {
         match self {
             ApiError::BadRequest(_) => (StatusCode::BAD_REQUEST, "bad_request"),
+            ApiError::SamplingContract { .. } => {
+                (StatusCode::BAD_REQUEST, "sampling_contract_violation")
+            }
             ApiError::ModelNotFound(_) => (StatusCode::NOT_FOUND, "model_not_found"),
             ApiError::UpstreamUnreachable { .. } => {
                 (StatusCode::BAD_GATEWAY, "upstream_unreachable")
@@ -242,7 +260,9 @@ impl IntoResponse for ApiError {
                 );
                 "service unavailable".to_string()
             }
-            ApiError::BadRequest(_) | ApiError::ModelNotFound(_) => self.to_string(),
+            ApiError::BadRequest(_)
+            | ApiError::ModelNotFound(_)
+            | ApiError::SamplingContract { .. } => self.to_string(),
         };
         let mut resp = (
             status,
@@ -442,4 +462,27 @@ mod tests {
             "ApiError::Internal must not leak anyhow chain to client; got: {body_str}"
         );
     }
+    /// A sampling-contract rejection must not be filed under `bad_request`:
+    /// an operator rolling `--sampling-param-conflict reject` across a fleet
+    /// has to be able to alert on contract rejections without them being
+    /// indistinguishable from clients sending malformed JSON.
+    #[test]
+    fn sampling_contract_has_a_distinct_code_from_other_bad_requests() {
+        let err = ApiError::SamplingContract {
+            param: "temperature",
+            detail: "got 0.5, expected 1 (or omit the field)".into(),
+        };
+        let (status, code_header, env) = parse_envelope(err.into_response());
+        assert_eq!(status, StatusCode::BAD_REQUEST);
+        assert_eq!(code_header.as_deref(), Some("sampling_contract_violation"));
+        assert_eq!(env.error.code, "sampling_contract_violation");
+        assert_ne!(env.error.code, "bad_request");
+        // The parameter and both values reach the client: a 400 here is
+        // actionable without an operator explaining it.
+        assert!(
+            env.error.message.contains("temperature") && env.error.message.contains("0.5"),
+            "got: {}",
+            env.error.message
+        );
+    }
 }

experimental/sgl-router/src/server/metrics.rs

@@ -42,6 +42,7 @@
 //! | `sgl_router_cache_aware_decisions_total` | Counter | `model_id`, `decision` |
 //! | `sgl_router_diverted_overlap_blocks` | Histogram | `model_id` |
 //! | `sgl_router_ingress_tokenize_errors_total` | Counter | `model_id` |
+//! | `sgl_router_sampling_contract_rejections_total` | Counter | `param` |
 //!
 //! `sgl_router_cache_aware_decisions_total` records exactly one decision per
 //! cache-aware prefill selection that resolves a worker, so the labels sum to
@@ -346,6 +347,7 @@ pub struct MetricsRegistry {
     cache_aware_decisions_total: Mutex<HashMap<CacheAwareDecisionKey, Arc<AtomicU64>>>,
     diverted_overlap_blocks: Mutex<HashMap<String, Histogram>>,
     ingress_tokenize_errors_total: Mutex<HashMap<String, Arc<AtomicU64>>>,
+    sampling_contract_rejections_total: Mutex<HashMap<&'static str, Arc<AtomicU64>>>,
 }
 
 #[derive(Debug, Hash, Eq, PartialEq, Clone)]
@@ -745,6 +747,24 @@ impl MetricsRegistry {
         counter.fetch_add(1, Ordering::Relaxed);
     }
 
+    /// Bump `sgl_router_sampling_contract_rejections_total{param}`.
+    ///
+    /// Recorded when the fleet-wide sampling contract refuses a request under
+    /// `--sampling-param-conflict reject`. This is the rollout gauge for the
+    /// flag: it answers "how much client traffic is the contract turning away,
+    /// and on which parameter" — which is otherwise unanswerable, because the
+    /// rejection reaches the client as a 400 like any other. `param` is a
+    /// wire name from a fixed enum, so the label set is bounded.
+    pub fn record_sampling_contract_rejection(&self, param: &'static str) {
+        let mut guard = self.sampling_contract_rejections_total.lock();
+        let counter = guard
+            .entry(param)
+            .or_insert_with(|| Arc::new(AtomicU64::new(0)))
+            .clone();
+        drop(guard);
+        counter.fetch_add(1, Ordering::Relaxed);
+    }
+
     /// Render the registry as a Prometheus 0.0.4 exposition-format string
     /// with no live worker snapshot. The per-worker gauges emit only their
     /// HELP/TYPE headers and a zeroed pool-size series. Production scrapes
@@ -1188,6 +1208,26 @@ impl MetricsRegistry {
         }
         drop(guard);
 
+        // sampling_contract_rejections_total
+        out.push_str(
+            "# HELP sgl_router_sampling_contract_rejections_total Requests refused by the fleet-wide sampling contract (--override-sampling-params under --sampling-param-conflict reject), by parameter.\n",
+        );
+        out.push_str("# TYPE sgl_router_sampling_contract_rejections_total counter\n");
+        let guard = self.sampling_contract_rejections_total.lock();
+        let mut entries: Vec<(&str, u64)> = guard
+            .iter()
+            .map(|(k, v)| (*k, v.load(Ordering::Relaxed)))
+            .collect();
+        entries.sort_by_key(|entry| entry.0);
+        for (param, value) in entries {
+            out.push_str(&format!(
+                "sgl_router_sampling_contract_rejections_total{{param=\"{}\"}} {}\n",
+                escape_label(param),
+                value,
+            ));
+        }
+        drop(guard);
+
         out
     }
 }
@@ -1739,4 +1779,29 @@ mod tests {
             "render did not escape backslash; got:\n{out}",
         );
     }
+    /// The contract's rollout gauge: absent until a request is actually
+    /// refused, then keyed by the parameter that refused it.
+    #[test]
+    fn sampling_contract_rejections_are_keyed_by_param() {
+        let reg = MetricsRegistry::new();
+        let out = reg.render();
+        assert!(out.contains("# TYPE sgl_router_sampling_contract_rejections_total counter"));
+        assert!(
+            !out.contains("sgl_router_sampling_contract_rejections_total{"),
+            "must emit no series before the first rejection"
+        );
+
+        reg.record_sampling_contract_rejection("temperature");
+        reg.record_sampling_contract_rejection("temperature");
+        reg.record_sampling_contract_rejection("top_p");
+        let out = reg.render();
+        assert!(
+            out.contains(r#"sgl_router_sampling_contract_rejections_total{param="temperature"} 2"#),
+            "got:\n{out}"
+        );
+        assert!(
+            out.contains(r#"sgl_router_sampling_contract_rejections_total{param="top_p"} 1"#),
+            "got:\n{out}"
+        );
+    }
 }