Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
104 changes: 91 additions & 13 deletions crates/libsy/src/algorithms/util/llm_judge.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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 {
Expand Down Expand Up @@ -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,
Expand All @@ -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(
Expand All @@ -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
Expand Down Expand Up @@ -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();
Expand Down
14 changes: 14 additions & 0 deletions crates/libsy/src/observability.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down
173 changes: 145 additions & 28 deletions crates/libsy/tests/observability.rs
Original file line number Diff line number Diff line change
Expand Up @@ -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(
Expand Down Expand Up @@ -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()
Expand Down Expand Up @@ -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()),
Expand Down Expand Up @@ -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(())
}
5 changes: 5 additions & 0 deletions crates/switchyard-server/README.md
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
Loading
Loading