feat(libsy): count classifier fail-open fallbacks on /metrics (#205)

Signed-off-by: Elyas Mehtabuddin <emehtabuddin@nvidia.com>

Author: elyasmnvidian

crates/libsy/src/algorithms/util/llm_judge.rs

@@ -22,7 +22,7 @@ use crate::core::algorithm::{Driver, LlmTarget};
 use crate::core::classifier::{Classification, Classifier};
 use crate::core::state::State;
 use crate::{LibsyError, Result};
-use switchyard_protocol::{Context, Decision, Request, Response};
+use switchyard_protocol::{Context, Decision, LlmClientError, Request, Response};
 
 /// Builds the classifier-specific message view presented to a structured judge.
 pub(crate) trait ClassifierInput: Send + Sync {
@@ -215,7 +215,7 @@ where
     /// A judge is an optimization, not a dependency: failing the caller's request because the
     /// judge is down would be worse than routing without it, so every failure — transport,
     /// mid-stream, or unparseable reply — is logged and folded into `None` for the policy's
-    /// fail-closed branch. A closed driver stream is folded too; the algorithm's next driver
+    /// fallback branch. A closed driver stream is folded too; the algorithm's next driver
     /// call surfaces it, so nothing is masked.
     async fn verdict(
         &self,
@@ -224,14 +224,6 @@ where
         driver: &Driver,
     ) -> Option<J::Verdict> {
         let judge_model = self.target.semantic_name.as_str();
-        let warn = |error: &dyn std::fmt::Display| {
-            tracing::warn!(
-                target: "libsy",
-                judge_model,
-                error = %error,
-                "judge verdict unavailable; routing without one"
-            );
-        };
 
         let response = driver
             .call_llm_target(
@@ -243,21 +235,57 @@ where
                 }),
             )
             .await
-            .inspect_err(|error| warn(error))
+            .inspect_err(|error| report_fail_open(judge_model, error, libsy_error_reason(error)))
             .ok()?;
         let aggregate = response
             .llm_response
             .into_agg()
             .await
-            .inspect_err(|error| warn(error))
+            .inspect_err(|error| report_fail_open(judge_model, error, client_error_reason(error)))
             .ok()?;
         self.judge
             .parse(&aggregate)
-            .inspect_err(|error| warn(error))
+            .inspect_err(|error| report_fail_open(judge_model, error, "parse_error"))
             .ok()
     }
 }
 
+/// Logs and counts a judge failure with a bounded label that excludes message content.
+fn report_fail_open(judge_model: &str, error: &dyn std::fmt::Display, reason: &'static str) {
+    tracing::warn!(
+        target: "libsy",
+        judge_model,
+        reason,
+        error = %error,
+        "judge verdict unavailable; routing without one"
+    );
+    crate::observability::record_classifier_fail_open(judge_model, reason);
+}
+
+/// Returns a bounded reason for a judge call that failed at the libsy layer.
+fn libsy_error_reason(error: &LibsyError) -> &'static str {
+    match error {
+        LibsyError::ClientCall { source, .. } => client_error_reason(source),
+        _ => "call_error",
+    }
+}
+
+/// Returns a bounded reason from the error kind and HTTP status only.
+fn client_error_reason(error: &LlmClientError) -> &'static str {
+    match error {
+        LlmClientError::Timeout { .. } => "timeout",
+        LlmClientError::Transport { .. } => "transport",
+        LlmClientError::UpstreamHttp { status, .. } if (500..=599).contains(status) => {
+            "upstream_5xx"
+        }
+        LlmClientError::UpstreamHttp { .. } => "upstream_non_5xx",
+        LlmClientError::InvalidResponse { .. } | LlmClientError::ResponseTranslation(_) => {
+            "invalid_response"
+        }
+        _ => "client_error",
+    }
+}
+
 #[async_trait]
 impl<J, P> Classifier<State> for JudgeClassifier<J, P>
 where
@@ -544,6 +572,56 @@ mod tests {
         Ok(())
     }
 
+    #[test]
+    fn client_errors_map_to_bounded_fail_open_reasons() {
+        let cases = vec![
+            (
+                LlmClientError::Timeout {
+                    source: "deadline exceeded".into(),
+                },
+                "timeout",
+            ),
+            (
+                LlmClientError::Transport {
+                    source: "connection refused".into(),
+                },
+                "transport",
+            ),
+            (
+                LlmClientError::UpstreamHttp {
+                    status: 500,
+                    body: "server error".to_string(),
+                },
+                "upstream_5xx",
+            ),
+            (
+                LlmClientError::UpstreamHttp {
+                    status: 302,
+                    body: "redirect".to_string(),
+                },
+                "upstream_non_5xx",
+            ),
+            (
+                LlmClientError::InvalidResponse {
+                    source: "invalid JSON".into(),
+                },
+                "invalid_response",
+            ),
+            (
+                LlmClientError::General("unexpected client failure".to_string()),
+                "client_error",
+            ),
+        ];
+        for (error, expected) in cases {
+            assert_eq!(client_error_reason(&error), expected);
+        }
+
+        let error = LibsyError::AlgorithmError {
+            message: "driver failed".to_string(),
+        };
+        assert_eq!(libsy_error_reason(&error), "call_error");
+    }
+
     #[tokio::test]
     async fn a_missing_driver_is_an_error_not_a_fallback() -> Result<()> {
         let mut request = request();

crates/libsy/src/observability.rs

@@ -513,6 +513,20 @@ fn record_routing_overhead(
     Some(overhead)
 }
 
+/// Records a judge failure that made the classifier route without a verdict.
+pub(crate) fn record_classifier_fail_open(judge_model: &str, reason: &'static str) {
+    meter()
+        .u64_counter("switchyard.classifier_fail_open")
+        .build()
+        .add(
+            1,
+            &[
+                KeyValue::new("judge_model", judge_model.to_string()),
+                KeyValue::new("reason", reason),
+            ],
+        );
+}
+
 /// Records the resolution of one offloaded model call: the call counter and
 /// latency histogram, the `outcome`/`error`/token fields on `span`, and a warn
 /// log when the call failed.

crates/libsy/tests/observability.rs

@@ -369,6 +369,58 @@ impl RoutedLlmClient for ClassifierClient {
     }
 }
 
+enum JudgeOutcome {
+    CallFailure,
+    Reply(&'static str),
+    StreamDecodeFailure,
+}
+
+/// Returns one configured judge outcome and serves the selected target normally.
+struct JudgeClient {
+    outcome: JudgeOutcome,
+}
+
+#[async_trait]
+impl RoutedLlmClient for JudgeClient {
+    async fn call(
+        &self,
+        _ctx: Context,
+        _request: Request,
+        decision: Arc<dyn Decision>,
+    ) -> Result<Response, LlmClientError> {
+        if decision.is_routed_call() {
+            return Ok(Response {
+                llm_response: LlmResponse::Agg(text_response(
+                    Some(decision.selected_model().to_string()),
+                    "routed response",
+                )),
+                metadata: None,
+            });
+        }
+        match &self.outcome {
+            JudgeOutcome::CallFailure => Err(LlmClientError::UpstreamHttp {
+                status: 500,
+                body: "server error".to_string(),
+            }),
+            JudgeOutcome::Reply(text) => Ok(Response {
+                llm_response: LlmResponse::Agg(text_response(None, *text)),
+                metadata: None,
+            }),
+            JudgeOutcome::StreamDecodeFailure => Ok(Response {
+                llm_response: LlmResponse::Stream(
+                    futures::stream::iter([Ok(LlmResponseStreamEvent::new(vec![
+                        LlmResponseChunk::DecodeError {
+                            message: "bad judge chunk".to_string(),
+                        },
+                    ]))])
+                    .boxed(),
+                ),
+                metadata: None,
+            }),
+        }
+    }
+}
+
 #[async_trait]
 impl RoutedLlmClient for UsageClient {
     async fn call(
@@ -453,6 +505,38 @@ fn algo(name: &str, model: &str, client: Option<Arc<dyn RoutedLlmClient>>) -> Ar
     })
 }
 
+fn classifier_router(
+    judge_model: &str,
+    efficient_model: &str,
+    capable_model: &str,
+    client: Arc<dyn RoutedLlmClient>,
+) -> switchyard_libsy::Result<Arc<dyn Algorithm>> {
+    let target = |name: &str| LlmTarget {
+        semantic_name: name.to_string(),
+        llm_client: Some(client.clone()),
+    };
+    let targets = LlmTargetSet::new(vec![target(efficient_model), target(capable_model)]);
+    Ok(Arc::new(LlmTaskClassifier::new(
+        LlmClassifierConfig::Capability {
+            judge_target: target(judge_model),
+            efficient_target: targets.get_target(efficient_model)?,
+            capable_target: targets.get_target(capable_model)?,
+            config: TaskClassifierConfig {
+                base_threshold: 0.5,
+                ..TaskClassifierConfig::default()
+            },
+        },
+    )?))
+}
+
+fn classifier_request() -> Request {
+    Request {
+        llm_request: text_request(Some("auto".to_string()), "classify this"),
+        raw_request: None,
+        metadata: None,
+    }
+}
+
 fn find_span(spans: &[SpanRecord], name: &str, field: &str, value: &str) -> SpanRecord {
     match spans
         .iter()
@@ -1032,34 +1116,10 @@ async fn classifier_metrics_count_only_the_final_routed_call() -> switchyard_lib
     let client = Arc::new(ClassifierClient {
         classifier_delay: Duration::from_millis(60),
         routed_delay: Duration::from_millis(200),
-    });
-    let target = |name: &str| LlmTarget {
-        semantic_name: name.to_string(),
-        llm_client: Some(client.clone()),
-    };
-    let targets = LlmTargetSet::new(vec![target("weak"), target("strong")]);
-    let weak = targets.get_target("weak")?;
-    let strong = targets.get_target("strong")?;
-    let router = Arc::new(LlmTaskClassifier::new(LlmClassifierConfig::Capability {
-        judge_target: target("classifier"),
-        efficient_target: weak,
-        capable_target: strong,
-        config: TaskClassifierConfig {
-            base_threshold: 0.5,
-            ..TaskClassifierConfig::default()
-        },
-    })?);
-
-    let (trace, _response) = router
-        .run(
-            Context::default(),
-            Request {
-                llm_request: text_request(Some("auto".to_string()), "classify this"),
-                raw_request: None,
-                metadata: None,
-            },
-        )
-        .await?;
+    }) as Arc<dyn RoutedLlmClient>;
+    let router = classifier_router("classifier", "weak", "strong", client)?;
+
+    let (trace, _response) = router.run(Context::default(), classifier_request()).await?;
 
     assert_eq!(
         trace.last().and_then(|decision| decision.routing_tier()),
@@ -1121,3 +1181,60 @@ async fn classifier_metrics_count_only_the_final_routed_call() -> switchyard_lib
     );
     Ok(())
 }
+
+#[tokio::test]
+async fn classifier_fail_open_records_each_failure_stage() -> switchyard_libsy::Result<()> {
+    let _guard = serialize_test().lock().await;
+    let (_store, exporter, provider, _, _) = telemetry();
+
+    let cases = [
+        ("fo-call", JudgeOutcome::CallFailure, Some("upstream_5xx")),
+        (
+            "fo-parse",
+            JudgeOutcome::Reply("not json at all"),
+            Some("parse_error"),
+        ),
+        (
+            "fo-stream-decode",
+            JudgeOutcome::StreamDecodeFailure,
+            Some("invalid_response"),
+        ),
+        (
+            "fo-valid",
+            JudgeOutcome::Reply(
+                r#"{"recommended_route":"strong","p_solve":0.3,"confidence":0.9,"abstain":false,"capability_boundary":"supported","primary_rule":"CAP-1","crux":"hard task"}"#,
+            ),
+            None,
+        ),
+    ];
+
+    for (judge_model, outcome, expected_reason) in cases {
+        let client = Arc::new(JudgeClient { outcome }) as Arc<dyn RoutedLlmClient>;
+        classifier_router(judge_model, "fo-weak", "fo-strong", client)?
+            .run(Context::default(), classifier_request())
+            .await?;
+
+        let snapshots = flushed_metrics(exporter, provider);
+        match expected_reason {
+            Some(reason) => assert_eq!(
+                u64_counter_value(
+                    &snapshots,
+                    "switchyard.classifier_fail_open",
+                    &[("reason", reason), ("judge_model", judge_model)],
+                ),
+                Some(1),
+                "case {reason} did not count the fail-open"
+            ),
+            None => assert_eq!(
+                u64_counter_value(
+                    &snapshots,
+                    "switchyard.classifier_fail_open",
+                    &[("judge_model", judge_model)],
+                ),
+                None,
+                "a valid verdict was counted as a fail-open"
+            ),
+        }
+    }
+    Ok(())
+}

crates/switchyard-server/README.md

@@ -148,13 +148,18 @@ Routed-call compatibility metrics are:
 | `switchyard_reasoning_tokens_total` | counter | `model`, optional `tier` | Reasoning output tokens |
 | `switchyard_total_latency_ms` | histogram | `model`, optional `tier` | Full-turn latency for successful routed responses |
 | `switchyard_routing_overhead_ms` | histogram | `algorithm` | Algorithm run time minus the call that served it |
+| `switchyard_classifier_fail_open_total` | counter | `judge_model`, `reason` | Judge failures that made a classifier route without a verdict |
 | `switchyard_client_responses_total` | counter | `outcome` | Final LLM-route responses |
 | `switchyard_upstream_attempts_total` | counter | `outcome`, `code` | Actual upstream HTTP attempts |
 | `switchyard_router_retry_recovered_total` | counter | none | Retry recoveries (currently always zero) |
 
 The `tier` label is `strong` or `weak` for a distinguishable built-in LLM-classifier decision and
 is omitted for untiered algorithms. Classifier calls are excluded from these families.
 
+`switchyard_classifier_fail_open_total` counts requests that still reached a target after the
+judge call failed. `judge_model` names the configured judge target, and `reason` is one of eight
+fixed error categories.
+
 `switchyard_total_latency_ms` observes an aggregate when it becomes available or a stream when it
 ends cleanly. Its clock starts in a router-wide middleware, before the request body is read and
 decoded, so it covers the same span as the Python server's request-ingress-to-completion