From 918da8a6c712f698ed0f2ee0f8d88e00907b11b8 Mon Sep 17 00:00:00 2001 From: ayushag Date: Fri, 16 Oct 2026 10:07:11 -0700 Subject: [PATCH 1/5] feat(composite): support custom-schema classifiers Signed-off-by: ayushag --- crates/libsy/src/algorithms/composite.rs | 43 +++- crates/libsy/src/algorithms/llm_class.rs | 55 +++-- crates/switchyard-runner/src/algorithm.rs | 68 +++++- .../tests/custom_composite.rs | 220 ++++++++++++++++++ docs/routing_algorithms/composite_routing.md | 25 ++ 5 files changed, 380 insertions(+), 31 deletions(-) create mode 100644 crates/switchyard-runner/tests/custom_composite.rs diff --git a/crates/libsy/src/algorithms/composite.rs b/crates/libsy/src/algorithms/composite.rs index 12ddfb39a8..b5d5359ee4 100644 --- a/crates/libsy/src/algorithms/composite.rs +++ b/crates/libsy/src/algorithms/composite.rs @@ -13,7 +13,9 @@ use async_trait::async_trait; use parking_lot::Mutex; use super::fall_through::FallThrough; -use super::llm_class::{LlmClassifierConfig, LlmTaskClassifier, TaskClassifierConfig}; +use super::llm_class::{ + CustomClassifierConfig, LlmClassifierConfig, LlmTaskClassifier, TaskClassifierConfig, +}; use super::stage::{StageRouterConfig, build_stage_route}; use super::util::affinity::{ClassifyTrigger, evict_if_full, has_new_user_turn, retention_key}; use super::util::stage::{Tier, set_fall_open}; @@ -142,13 +144,48 @@ impl CompositeRouter { let judge = LlmTaskClassifier::new(LlmClassifierConfig::Capability { config: judge_config, })?; + Self::with_judge( + Arc::new(judge), + trigger, + message_hash_fallback, + config.stage, + ) + } + + /// Uses a custom-schema judge selecting the `efficient` or `capable` category. + /// Other categories abstain, retaining the previous tier. The stage router + /// continues routing within the turn. `fail_open` recovers from client failures. + pub fn with_custom_classifier( + judge: CustomClassifierConfig, + stage: StageRouterConfig, + fail_open: bool, + ) -> Result { + if judge.classify_trigger == ClassifyTrigger::EveryRequest { + return Err(LibsyError::AlgorithmError { + message: "composite: classify_trigger must be user_turn or new_session".to_string(), + }); + } + Self::with_judge( + LlmTaskClassifier::custom_classifier(&judge, fail_open)?, + judge.classify_trigger, + judge.message_hash_fallback, + stage, + ) + } + + fn with_judge( + judge: Arc>, + trigger: ClassifyTrigger, + message_hash_fallback: bool, + stage: StageRouterConfig, + ) -> Result { let setter = TierSetter { - judge: Arc::new(judge), + judge, trigger, message_hash_fallback, tiers: Mutex::new(HashMap::new()), }; - let route = build_stage_route(config.stage)? + let route = build_stage_route(stage)? .with_name(COMPOSITE) .with_processor(Arc::new(setter)); Ok(Self { route }) diff --git a/crates/libsy/src/algorithms/llm_class.rs b/crates/libsy/src/algorithms/llm_class.rs index a956837902..5aa948101e 100644 --- a/crates/libsy/src/algorithms/llm_class.rs +++ b/crates/libsy/src/algorithms/llm_class.rs @@ -720,40 +720,49 @@ impl LlmTaskClassifier { } fn build_custom(default_target: Category, config: CustomClassifierConfig) -> Result { + Self::from_classifier( + Self::custom_classifier(&config, false)?, + ClassifierRouteConfig { + default_target, + classify_trigger: config.classify_trigger, + message_hash_fallback: config.message_hash_fallback, + }, + ) + } + + /// The judge face alone, for composite's own trigger and retained tier. + pub(crate) fn custom_classifier( + config: &CustomClassifierConfig, + fail_open: bool, + ) -> Result>> { config.validate()?; - let CustomClassifierConfig { - prompt, - response_schema, - policy, - classify_trigger, - message_hash_fallback, - recent_turn_window, - max_output_tokens, - } = config; - let contract = ClassifierContract::from_inner_schema(&prompt, response_schema)?; - let policy = match policy { + let contract = + ClassifierContract::from_inner_schema(&config.prompt, config.response_schema.clone())?; + let policy = match &config.policy { CustomClassifierPolicy::TargetSelector { selector } => { CustomPolicyRuntime::TargetSelector(TargetSelectorPolicy::new(selector)?) } }; - let classifier: Arc> = Arc::new(JudgeClassifier::new( + let classifier = JudgeClassifier::new( StructuredJudge::new( - TaskInput { recent_turn_window }, + TaskInput { + recent_turn_window: config.recent_turn_window, + }, contract, JsonSchemaDecoder::new(), - JudgeRuntimeConfig::new(max_output_tokens)?, + JudgeRuntimeConfig::new(config.max_output_tokens)?, ), policy, - )); - - Self::from_classifier( - classifier, - ClassifierRouteConfig { - default_target, - classify_trigger, - message_hash_fallback, - }, ) + .with_error_recovery(fail_open); + let classifier = if fail_open { + classifier.with_evidence(|_, verdict| { + verdict.map(|_| serde_json::json!({"source": "llm-classifier"})) + }) + } else { + classifier + }; + Ok(Arc::new(classifier)) } fn build_escalation( diff --git a/crates/switchyard-runner/src/algorithm.rs b/crates/switchyard-runner/src/algorithm.rs index a994ffc9c6..12ef5c961c 100644 --- a/crates/switchyard-runner/src/algorithm.rs +++ b/crates/switchyard-runner/src/algorithm.rs @@ -574,6 +574,13 @@ pub enum AdvisorTriggerConfig { pub struct StageClassifierConfig { /// Target the judge is called through. Not a routing destination. pub target: String, + /// Composite accepts custom-schema judges as well as capability judges. + #[serde(default)] + pub mode: Option, + #[serde(default)] + pub response_schema: Option, + #[serde(default)] + pub policy: Option, #[serde(default)] pub decision: Option, #[serde(default)] @@ -645,6 +652,16 @@ impl StageClassifierConfig { tiers: &StageTierConfig, targets: &BTreeMap, ) -> AlgorithmResult { + if self + .mode + .is_some_and(|mode| !matches!(mode, ClassifierMode::Capability)) + || self.response_schema.is_some() + || self.policy.is_some() + { + return Err(AlgorithmConfigError::new( + "custom classifier settings require composite with mode = custom", + )); + } let config = LlmClassifierRouteConfig { classifier_target: self.target.clone(), strong_target: Some(tiers.capable_target.clone()), @@ -668,6 +685,39 @@ impl StageClassifierConfig { _ => Err(AlgorithmConfigError::new("expected a capability judge")), } } + + fn custom_classifier_config( + &self, + route_name: &str, + ) -> AlgorithmResult { + if self.decision.is_some() + || self.base_threshold.is_some() + || self.threshold_step.is_some() + || self.response_format_type != ClassifierResponseFormat::JsonSchema + { + return Err(AlgorithmConfigError::new( + "composite custom classifier cannot use capability thresholds, decision settings, or response_format_type", + )); + } + let schema = + required_classifier_field(route_name, "response_schema", &self.response_schema)?; + let schema = serde_json::from_str(&schema).map_err(|error| { + AlgorithmConfigError::with_source( + format!("composite route {route_name}: response_schema is invalid JSON"), + error, + ) + })?; + let mut config = CustomClassifierConfig::new( + required_classifier_field(route_name, "prompt", &self.prompt)?, + schema, + required_classifier_field(route_name, "policy", &self.policy)?.into_libsy(), + ); + config.classify_trigger = self.classify_trigger; + config.message_hash_fallback = self.message_hash_fallback; + config.recent_turn_window = self.recent_turn_window; + config.max_output_tokens = self.max_output_tokens; + Ok(config) + } } impl AlgorithmSpec { @@ -1600,11 +1650,19 @@ fn build_algorithm( } stage_config.tool_semantics = stage.tool_semantics.clone(); stage_config.handoff_notes = stage.handoff_notes.clone(); - let config = CompositeRouterConfig { - judge: classifier.task_classifier_config(route_name, stage, targets)?, - stage: stage_config, - }; - let algorithm = CompositeRouter::new(config).map_err(|error| { + let algorithm = if matches!(classifier.mode, Some(ClassifierMode::Custom)) { + CompositeRouter::with_custom_classifier( + classifier.custom_classifier_config(route_name)?, + stage_config, + classifier.fail_open.unwrap_or(true), + ) + } else { + CompositeRouter::new(CompositeRouterConfig { + judge: classifier.task_classifier_config(route_name, stage, targets)?, + stage: stage_config, + }) + } + .map_err(|error| { AlgorithmConfigError::with_source( format!("composite route {route_name}: {error}"), error, diff --git a/crates/switchyard-runner/tests/custom_composite.rs b/crates/switchyard-runner/tests/custom_composite.rs new file mode 100644 index 0000000000..91082535d6 --- /dev/null +++ b/crates/switchyard-runner/tests/custom_composite.rs @@ -0,0 +1,220 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +use async_trait::async_trait; +use std::collections::BTreeMap; +use std::sync::{ + Arc, Mutex, + atomic::{AtomicUsize, Ordering}, +}; +use switchyard_llm_client::ClientRouter; +use switchyard_protocol::{ + LlmClientError, LlmRequest, LlmResponse, Message, Metadata, Request, Response, Role, + RoutedLlmClient, text_response, +}; +use switchyard_runner::{AlgorithmSpec, ModelCapabilities, Route, Runner}; + +const CONFIG: &str = r#" +schema_version = 1 +[llm_clients.local] +format = "openai_chat" +base_url = "http://127.0.0.1:1/v1" +[targets.fast] +id = "fast-model" +llm_client = "local" +[targets.strong] +id = "strong-model" +llm_client = "local" +[routes.composed] +id = "composite-id" +type = "composite" +[routes.composed.classifier] +target = "fast" +mode = "custom" +prompt = 'Choose efficient or capable for the current task.' +response_schema = '{"type":"object","properties":{"decision":{"type":"object","properties":{"target":{"type":"string","enum":["efficient","capable"]}},"required":["target"],"additionalProperties":false}},"required":["decision"],"additionalProperties":false}' +policy = { type = "target_selector", selector = "/decision/target" } +classify_trigger = "user_turn" +max_output_tokens = 384 +[routes.composed.stage] +efficient_target = "fast" +capable_target = "strong" +"#; + +struct CustomJudge { + reply: Mutex>, + calls: AtomicUsize, +} + +#[async_trait] +impl RoutedLlmClient for CustomJudge { + async fn call(&self, request: Request) -> Result { + self.calls.fetch_add(1, Ordering::SeqCst); + assert!( + request.llm_request.instructions[0] + .content + .iter() + .any(|block| { + matches!(block, switchyard_protocol::ContentBlock::Text { text } + if text == "Choose efficient or capable for the current task.") + }) + ); + let reply = self + .reply + .lock() + .unwrap() + .ok_or_else(|| LlmClientError::General("judge unavailable".into()))?; + Ok(Response { + llm_response: LlmResponse::Agg(text_response(None, reply)), + metadata: None, + upstream_headers: Default::default(), + }) + } +} + +fn custom_judge(reply: Option<&'static str>) -> Arc { + Arc::new(CustomJudge { + reply: Mutex::new(reply), + calls: AtomicUsize::new(0), + }) +} + +fn custom_route(config: &str, judge: Arc) -> Route { + let mut config: toml::Value = toml::from_str(config).unwrap(); + config["routes"]["composed"] + .as_table_mut() + .unwrap() + .remove("id"); + let spec: AlgorithmSpec = config["routes"]["composed"].clone().try_into().unwrap(); + let targets = BTreeMap::from([ + ("fast".into(), "fast-model".into()), + ("strong".into(), "strong-model".into()), + ]); + let (algorithm, models) = spec + .build_with_runtime_models("composed", &targets) + .unwrap(); + Route::new( + algorithm, + ClientRouter::single(judge), + None, + ModelCapabilities::default(), + None, + None, + Vec::new(), + models, + ) +} + +fn request() -> Request { + Request { + llm_request: LlmRequest { + messages: vec![Message::text(Role::User, "Analyze a difficult task")], + ..Default::default() + }, + metadata: Some(Metadata { + session_id: Some("a".into()), + ..Default::default() + }), + ..Default::default() + } +} + +#[tokio::test] +async fn custom_composite_preserves_tier_across_continuations_and_judge_failure() { + let judge = custom_judge(Some(r#"{"decision":{"target":"capable"}}"#)); + let route = custom_route(CONFIG, judge.clone()); + let mut request = request(); + let first = route.decide(request.clone()).await.unwrap(); + assert_eq!(first.selected_model_id().unwrap().as_str(), "strong-model"); + let mut continuation = request.clone(); + continuation + .llm_request + .messages + .push(Message::text(Role::Assistant, "Working")); + let outcome = route.decide(continuation).await.unwrap(); + assert_eq!( + outcome.selected_model_id().unwrap().as_str(), + "strong-model" + ); + assert_eq!(judge.calls.load(Ordering::SeqCst), 1); + + for reply in [ + None, + Some("not JSON"), + Some("{}"), + Some(r#"{"decision":{"target":"judge"}}"#), + Some(r#"{"decision":{"target":"other"}}"#), + ] { + *judge.reply.lock().unwrap() = reply; + let outcome = route.decide(request.clone()).await.unwrap(); + assert_eq!( + outcome.selected_model_id().unwrap().as_str(), + "strong-model" + ); + assert_eq!( + outcome.metadata.unwrap().evidence.unwrap()["source"], + "fail_open" + ); + } + *judge.reply.lock().unwrap() = None; + request.metadata.as_mut().unwrap().session_id = Some("b".into()); + let outcome = route.decide(request.clone()).await.unwrap(); + assert_eq!(outcome.selected_model_id().unwrap().as_str(), "fast-model"); + request.metadata.as_mut().unwrap().session_id = Some("a".into()); + *judge.reply.lock().unwrap() = Some(r#"{"decision":{"target":"efficient"}}"#); + let outcome = route.decide(request.clone()).await.unwrap(); + let mut expected = request.llm_request; + expected.model = Some("fast-model".into()); + assert_eq!(outcome.request.llm_request, expected); + assert!(outcome.response.is_none()); +} + +#[tokio::test] +async fn custom_composite_respects_new_session_and_fail_closed() { + let config = CONFIG.replace( + "classify_trigger = \"user_turn\"", + "classify_trigger = \"new_session\"", + ); + let judge = custom_judge(Some(r#"{"decision":{"target":"capable"}}"#)); + let route = custom_route(&config, judge.clone()); + for _ in 0..2 { + let outcome = route.decide(request()).await.unwrap(); + assert_eq!( + outcome.selected_model_id().unwrap().as_str(), + "strong-model" + ); + } + assert_eq!(judge.calls.load(Ordering::SeqCst), 1); + let closed = CONFIG.replace("mode = \"custom\"", "mode = \"custom\"\nfail_open = false"); + let route = custom_route(&closed, custom_judge(None)); + assert!(route.decide(request()).await.is_err()); +} + +#[test] +fn custom_composite_rejects_incompatible_or_missing_settings() { + let config = CONFIG; + assert!(Runner::from_toml(config).is_ok()); + for invalid in [ + config.replace( + "mode = \"custom\"", + "mode = \"custom\"\nbase_threshold = 0.5", + ), + config.replace("mode = \"custom\"", "mode = \"escalation\""), + config.replace("mode = \"custom\"", "mode = \"capability\""), + config.replace( + "prompt = 'Choose efficient or capable for the current task.'", + "", + ), + config.replace( + "selector = \"/decision/target\"", + "selector = \"/decision/target~2\"", + ), + config.replace( + "classify_trigger = \"user_turn\"", + "classify_trigger = \"every_request\"", + ), + config.replace("response_schema = '", "response_schema = 'not-json"), + ] { + assert!(Runner::from_toml(&invalid).is_err()); + } +} diff --git a/docs/routing_algorithms/composite_routing.md b/docs/routing_algorithms/composite_routing.md index 210736f8ec..4dac018f2e 100644 --- a/docs/routing_algorithms/composite_routing.md +++ b/docs/routing_algorithms/composite_routing.md @@ -59,6 +59,31 @@ nor a second `classifier`. The outer classifier supplies the default tier. Custom `tool_semantics` remain stage-local and have the same additive behavior as on a standalone stage route. +## Custom LLM classifier + +To select a tier directly with your own prompt, keep the stage block and use: + +```toml +[routes.switchyard.classifier] +mode = "custom" +target = "judge" +classify_trigger = "user_turn" +max_output_tokens = 64 +fail_open = true +prompt = 'Choose efficient unless capable offers a concrete advantage for this task. Return JSON with decision.target set to efficient or capable.' +response_schema = ''' +{"type":"object","properties":{"decision":{"type":"object","properties":{"target":{"type":"string","enum":["efficient","capable"]}},"required":["target"],"additionalProperties":false}},"required":["decision"],"additionalProperties":false} +''' +policy = { type = "target_selector", selector = "/decision/target" } +``` + +The schema is validated locally. The selector names a tier, and the existing +stage targets supply its model. With `fail_open = true` (the default), a missing +or invalid verdict or a client failure retains the previous tier, or uses efficient +for a new session. With `fail_open = false`, these errors propagate instead. +Capability thresholds and decision-model +settings cannot be combined with `mode = "custom"`. + ## Decision model classifier Use a decision target as the classifier. Keep the stage block above From d67935afe95c61f5b7c9d6476db38983a24ef2e4 Mon Sep 17 00:00:00 2001 From: ayushag Date: Fri, 16 Oct 2026 10:16:38 -0700 Subject: [PATCH 2/5] refactor(composite): unify classifier construction Signed-off-by: ayushag --- crates/libsy/src/algorithms/composite.rs | 124 +++++++++--------- crates/libsy/src/lib.rs | 2 +- crates/switchyard-runner/src/algorithm.rs | 37 +++--- .../tests/custom_composite.rs | 43 ++++++ docs/routing_algorithms/composite_routing.md | 6 +- 5 files changed, 129 insertions(+), 83 deletions(-) diff --git a/crates/libsy/src/algorithms/composite.rs b/crates/libsy/src/algorithms/composite.rs index b5d5359ee4..260e556483 100644 --- a/crates/libsy/src/algorithms/composite.rs +++ b/crates/libsy/src/algorithms/composite.rs @@ -105,10 +105,23 @@ impl Processor for TierSetter { } } +/// Classifier settings for the composite router's retained tier. +pub enum CompositeJudgeConfig { + /// Selects a tier from a capability verdict. + Capability(TaskClassifierConfig), + /// Selects a tier using a custom schema and policy. + Custom { + /// Custom classifier contract and trigger settings. + config: CustomClassifierConfig, + /// Recovers from client failures by retaining the previous tier. + fail_open: bool, + }, +} + /// A judge stacked over a stage router. pub struct CompositeRouterConfig { /// Judge settings, including how often `classify_trigger` runs it. - pub judge: TaskClassifierConfig, + pub judge: CompositeJudgeConfig, /// Serves the turns, with the tier the judge picked as its fall-open default. pub stage: StageRouterConfig, } @@ -126,66 +139,41 @@ impl CompositeRouter { /// A stage router carrying its own judge is allowed, but that judge sits ahead /// of the fall-open tier and so answers most of the turns this one set a tier for. pub fn new(config: CompositeRouterConfig) -> Result { - if config.judge.classify_trigger == ClassifyTrigger::EveryRequest { - return Err(LibsyError::AlgorithmError { - message: "composite: classify_trigger must be user_turn or new_session".to_string(), - }); - } - let trigger = config.judge.classify_trigger; - let message_hash_fallback = config.judge.message_hash_fallback; - // Only the judge's Classifier face is used, so its own affinity never runs. - // Leaving these set would apply the standalone route's pairing rules to a - // trigger this router implements itself. - let judge_config = TaskClassifierConfig { - classify_trigger: ClassifyTrigger::EveryRequest, - message_hash_fallback: false, - ..config.judge + let (trigger, message_hash_fallback) = match &config.judge { + CompositeJudgeConfig::Capability(judge) => { + (judge.classify_trigger, judge.message_hash_fallback) + } + CompositeJudgeConfig::Custom { config, .. } => { + (config.classify_trigger, config.message_hash_fallback) + } }; - let judge = LlmTaskClassifier::new(LlmClassifierConfig::Capability { - config: judge_config, - })?; - Self::with_judge( - Arc::new(judge), - trigger, - message_hash_fallback, - config.stage, - ) - } - - /// Uses a custom-schema judge selecting the `efficient` or `capable` category. - /// Other categories abstain, retaining the previous tier. The stage router - /// continues routing within the turn. `fail_open` recovers from client failures. - pub fn with_custom_classifier( - judge: CustomClassifierConfig, - stage: StageRouterConfig, - fail_open: bool, - ) -> Result { - if judge.classify_trigger == ClassifyTrigger::EveryRequest { + if trigger == ClassifyTrigger::EveryRequest { return Err(LibsyError::AlgorithmError { message: "composite: classify_trigger must be user_turn or new_session".to_string(), }); } - Self::with_judge( - LlmTaskClassifier::custom_classifier(&judge, fail_open)?, - judge.classify_trigger, - judge.message_hash_fallback, - stage, - ) - } - - fn with_judge( - judge: Arc>, - trigger: ClassifyTrigger, - message_hash_fallback: bool, - stage: StageRouterConfig, - ) -> Result { + let judge: Arc> = match config.judge { + CompositeJudgeConfig::Capability(judge) => { + // Composite owns the trigger and retention; standalone affinity never runs. + Arc::new(LlmTaskClassifier::new(LlmClassifierConfig::Capability { + config: TaskClassifierConfig { + classify_trigger: ClassifyTrigger::EveryRequest, + message_hash_fallback: false, + ..judge + }, + })?) + } + CompositeJudgeConfig::Custom { config, fail_open } => { + LlmTaskClassifier::custom_classifier(&config, fail_open)? + } + }; let setter = TierSetter { judge, trigger, message_hash_fallback, tiers: Mutex::new(HashMap::new()), }; - let route = build_stage_route(stage)? + let route = build_stage_route(config.stage)? .with_name(COMPOSITE) .with_processor(Arc::new(setter)); Ok(Self { route }) @@ -252,7 +240,7 @@ mod tests { fn hash_keyed_router() -> Result> { Ok(Arc::new(CompositeRouter::new(CompositeRouterConfig { - judge: TaskClassifierConfig { + judge: CompositeJudgeConfig::Capability(TaskClassifierConfig { judge: CapabilityJudgeConfig::Llm(LlmCapabilityConfig { base_threshold: 0.5, ..LlmCapabilityConfig::default() @@ -260,35 +248,47 @@ mod tests { classify_trigger: ClassifyTrigger::UserTurn, message_hash_fallback: true, ..Default::default() - }, + }), stage: StageRouterConfig::new(PickerMode::EfficientFirst, 0.5), })?)) } fn router() -> Result> { Ok(Arc::new(CompositeRouter::new(CompositeRouterConfig { - judge: TaskClassifierConfig { + judge: CompositeJudgeConfig::Capability(TaskClassifierConfig { judge: CapabilityJudgeConfig::Llm(LlmCapabilityConfig { base_threshold: 0.5, ..LlmCapabilityConfig::default() }), classify_trigger: ClassifyTrigger::UserTurn, ..Default::default() - }, + }), stage: StageRouterConfig::new(PickerMode::EfficientFirst, 0.5), })?)) } #[test] fn rejects_every_request_as_a_trigger() { - let config = CompositeRouterConfig { - judge: TaskClassifierConfig::default(), - stage: StageRouterConfig::new(PickerMode::EfficientFirst, 0.5), - }; - assert!(matches!( - CompositeRouter::new(config), - Err(LibsyError::AlgorithmError { .. }) - )); + for judge in [ + CompositeJudgeConfig::Capability(TaskClassifierConfig::default()), + CompositeJudgeConfig::Custom { + config: CustomClassifierConfig::new( + "Choose a tier", + serde_json::json!({"type": "object"}), + crate::CustomClassifierPolicy::target_selector("/decision/target"), + ), + fail_open: true, + }, + ] { + let error = CompositeRouter::new(CompositeRouterConfig { + judge, + stage: StageRouterConfig::new(PickerMode::EfficientFirst, 0.5), + }) + .err() + .expect("every_request must be rejected for both classifier modes"); + assert!(matches!(error, LibsyError::AlgorithmError { message } + if message == "composite: classify_trigger must be user_turn or new_session")); + } } /// A route may point its judge at the same target it serves capable turns on. diff --git a/crates/libsy/src/lib.rs b/crates/libsy/src/lib.rs index efd61f63a9..0c091db879 100644 --- a/crates/libsy/src/lib.rs +++ b/crates/libsy/src/lib.rs @@ -19,7 +19,7 @@ pub use error::{DriverError, LibsyError, Result}; mod algorithms; pub use algorithms::advisor_gate::{AdvisorGate, AdvisorGateConfig, GateTrigger}; -pub use algorithms::composite::{CompositeRouter, CompositeRouterConfig}; +pub use algorithms::composite::{CompositeJudgeConfig, CompositeRouter, CompositeRouterConfig}; pub use algorithms::llm_class::{ CapabilityJudgeConfig, CustomClassifierConfig, CustomClassifierPolicy, DecisionJudgeConfig, LlmCapabilityConfig, LlmClassifierConfig, LlmTaskClassifier, TaskClassifierConfig, diff --git a/crates/switchyard-runner/src/algorithm.rs b/crates/switchyard-runner/src/algorithm.rs index 12ef5c961c..2bb1964d18 100644 --- a/crates/switchyard-runner/src/algorithm.rs +++ b/crates/switchyard-runner/src/algorithm.rs @@ -11,12 +11,13 @@ use std::sync::Arc; use libsy::{ AdvisorGate, AdvisorGateConfig, Algorithm, BreakerConfig, CapabilityJudgeConfig, - ClassifierContractConfig, ClassifierResponseFormat, ClassifyTrigger, CompositeRouter, - CompositeRouterConfig, CustomClassifierConfig, CustomClassifierPolicy, DecisionJudgeConfig, - EscalationJudgeConfig, GateTrigger, HandoffNoteConfig, LlmCapabilityConfig, - LlmClassifierConfig, LlmFallback, LlmTaskClassifier, Noop, Passthrough, PickerMode, - PlanExecute, PlanExecuteConfig, Random, RuntimeModels, StageRouter, StageRouterConfig, - SubagentRouter, SubagentRouterConfig, TaskClassifierConfig, ToolSemantics, Vgr, VgrConfig, + ClassifierContractConfig, ClassifierResponseFormat, ClassifyTrigger, CompositeJudgeConfig, + CompositeRouter, CompositeRouterConfig, CustomClassifierConfig, CustomClassifierPolicy, + DecisionJudgeConfig, EscalationJudgeConfig, GateTrigger, HandoffNoteConfig, + LlmCapabilityConfig, LlmClassifierConfig, LlmFallback, LlmTaskClassifier, Noop, Passthrough, + PickerMode, PlanExecute, PlanExecuteConfig, Random, RuntimeModels, StageRouter, + StageRouterConfig, SubagentRouter, SubagentRouterConfig, TaskClassifierConfig, ToolSemantics, + Vgr, VgrConfig, }; use serde::Deserialize; use serde_json::Value; @@ -1650,18 +1651,20 @@ fn build_algorithm( } stage_config.tool_semantics = stage.tool_semantics.clone(); stage_config.handoff_notes = stage.handoff_notes.clone(); - let algorithm = if matches!(classifier.mode, Some(ClassifierMode::Custom)) { - CompositeRouter::with_custom_classifier( - classifier.custom_classifier_config(route_name)?, - stage_config, - classifier.fail_open.unwrap_or(true), - ) + let judge = if matches!(classifier.mode, Some(ClassifierMode::Custom)) { + CompositeJudgeConfig::Custom { + config: classifier.custom_classifier_config(route_name)?, + fail_open: classifier.fail_open.unwrap_or(true), + } } else { - CompositeRouter::new(CompositeRouterConfig { - judge: classifier.task_classifier_config(route_name, stage, targets)?, - stage: stage_config, - }) - } + CompositeJudgeConfig::Capability( + classifier.task_classifier_config(route_name, stage, targets)?, + ) + }; + let algorithm = CompositeRouter::new(CompositeRouterConfig { + judge, + stage: stage_config, + }) .map_err(|error| { AlgorithmConfigError::with_source( format!("composite route {route_name}: {error}"), diff --git a/crates/switchyard-runner/tests/custom_composite.rs b/crates/switchyard-runner/tests/custom_composite.rs index 91082535d6..0ffb57bb04 100644 --- a/crates/switchyard-runner/tests/custom_composite.rs +++ b/crates/switchyard-runner/tests/custom_composite.rs @@ -188,6 +188,49 @@ async fn custom_composite_respects_new_session_and_fail_closed() { let closed = CONFIG.replace("mode = \"custom\"", "mode = \"custom\"\nfail_open = false"); let route = custom_route(&closed, custom_judge(None)); assert!(route.decide(request()).await.is_err()); + + let judge = custom_judge(Some(r#"{"decision":{"target":"capable"}}"#)); + let route = custom_route(&closed, judge.clone()); + route.decide(request()).await.unwrap(); + // Invalid verdicts abstain independently of client-error recovery. + for reply in ["not JSON", "{}", r#"{"decision":{"target":"other"}}"#] { + *judge.reply.lock().unwrap() = Some(reply); + let outcome = route.decide(request()).await.unwrap(); + assert_eq!( + outcome.selected_model_id().unwrap().as_str(), + "strong-model" + ); + assert_eq!( + outcome.metadata.unwrap().evidence.unwrap()["source"], + "retained" + ); + } +} + +#[tokio::test] +async fn custom_composite_retains_tier_with_message_hash_fallback() { + for trigger in ["user_turn", "new_session"] { + let config = CONFIG.replace("user_turn", trigger).replace( + "mode = \"custom\"", + "mode = \"custom\"\nmessage_hash_fallback = true", + ); + let judge = custom_judge(Some(r#"{"decision":{"target":"capable"}}"#)); + let route = custom_route(&config, judge.clone()); + let mut request = request(); + request.metadata = None; + for _ in 0..2 { + let outcome = route.decide(request.clone()).await.unwrap(); + assert_eq!( + outcome.selected_model_id().unwrap().as_str(), + "strong-model" + ); + request + .llm_request + .messages + .push(Message::text(Role::Assistant, "Working")); + } + assert_eq!(judge.calls.load(Ordering::SeqCst), 1); + } } #[test] diff --git a/docs/routing_algorithms/composite_routing.md b/docs/routing_algorithms/composite_routing.md index 4dac018f2e..09d9052aaa 100644 --- a/docs/routing_algorithms/composite_routing.md +++ b/docs/routing_algorithms/composite_routing.md @@ -78,9 +78,9 @@ policy = { type = "target_selector", selector = "/decision/target" } ``` The schema is validated locally. The selector names a tier, and the existing -stage targets supply its model. With `fail_open = true` (the default), a missing -or invalid verdict or a client failure retains the previous tier, or uses efficient -for a new session. With `fail_open = false`, these errors propagate instead. +stage targets supply its model. A missing or invalid verdict retains the previous +tier, or uses efficient for a new session. Client failures do the same with +`fail_open = true` (the default); with `fail_open = false`, client errors propagate. Capability thresholds and decision-model settings cannot be combined with `mode = "custom"`. From d43331cb0a983aec08ca1e3d3da8a098a8ca9041 Mon Sep 17 00:00:00 2001 From: ayushag Date: Fri, 16 Oct 2026 10:21:22 -0700 Subject: [PATCH 3/5] fix(composite): retain session fallback and reject invalid strict verdicts Signed-off-by: ayushag --- crates/libsy/src/algorithms/composite.rs | 22 ++- .../tests/custom_composite.rs | 155 ++++++++---------- docs/routing_algorithms/composite_routing.md | 8 +- 3 files changed, 95 insertions(+), 90 deletions(-) diff --git a/crates/libsy/src/algorithms/composite.rs b/crates/libsy/src/algorithms/composite.rs index 260e556483..cb120e05cd 100644 --- a/crates/libsy/src/algorithms/composite.rs +++ b/crates/libsy/src/algorithms/composite.rs @@ -37,6 +37,7 @@ struct TierSetter { judge: Arc>, trigger: ClassifyTrigger, message_hash_fallback: bool, + require_verdict: bool, tiers: Mutex>, } @@ -93,6 +94,17 @@ impl Processor for TierSetter { } return Ok(()); } + if self.require_verdict { + return Err(LibsyError::AlgorithmError { + message: "composite: custom classifier must select efficient or capable".into(), + }); + } + if self.trigger == ClassifyTrigger::NewSession + && let Some(identity) = identity.as_ref() + { + // A failed opening judgment is still the session's one judgment. + self.retain(identity.clone(), Tier::Efficient); + } } // Either not this request's turn or the judge had no verdict, so the last // tier stands rather than dropping back to the picker default. @@ -113,7 +125,7 @@ pub enum CompositeJudgeConfig { Custom { /// Custom classifier contract and trigger settings. config: CustomClassifierConfig, - /// Recovers from client failures by retaining the previous tier. + /// Retains the previous tier on failure; otherwise requires a valid tier verdict. fail_open: bool, }, } @@ -152,6 +164,13 @@ impl CompositeRouter { message: "composite: classify_trigger must be user_turn or new_session".to_string(), }); } + let require_verdict = matches!( + &config.judge, + CompositeJudgeConfig::Custom { + fail_open: false, + .. + } + ); let judge: Arc> = match config.judge { CompositeJudgeConfig::Capability(judge) => { // Composite owns the trigger and retention; standalone affinity never runs. @@ -171,6 +190,7 @@ impl CompositeRouter { judge, trigger, message_hash_fallback, + require_verdict, tiers: Mutex::new(HashMap::new()), }; let route = build_stage_route(config.stage)? diff --git a/crates/switchyard-runner/tests/custom_composite.rs b/crates/switchyard-runner/tests/custom_composite.rs index 0ffb57bb04..c599b00c61 100644 --- a/crates/switchyard-runner/tests/custom_composite.rs +++ b/crates/switchyard-runner/tests/custom_composite.rs @@ -9,8 +9,8 @@ use std::sync::{ }; use switchyard_llm_client::ClientRouter; use switchyard_protocol::{ - LlmClientError, LlmRequest, LlmResponse, Message, Metadata, Request, Response, Role, - RoutedLlmClient, text_response, + LlmClientError, LlmResponse, Message, Metadata, Request, Response, Role, RoutedLlmClient, + text_request, text_response, }; use switchyard_runner::{AlgorithmSpec, ModelCapabilities, Route, Runner}; @@ -48,17 +48,8 @@ struct CustomJudge { #[async_trait] impl RoutedLlmClient for CustomJudge { - async fn call(&self, request: Request) -> Result { + async fn call(&self, _request: Request) -> Result { self.calls.fetch_add(1, Ordering::SeqCst); - assert!( - request.llm_request.instructions[0] - .content - .iter() - .any(|block| { - matches!(block, switchyard_protocol::ContentBlock::Text { text } - if text == "Choose efficient or capable for the current task.") - }) - ); let reply = self .reply .lock() @@ -107,10 +98,7 @@ fn custom_route(config: &str, judge: Arc) -> Route { fn request() -> Request { Request { - llm_request: LlmRequest { - messages: vec![Message::text(Role::User, "Analyze a difficult task")], - ..Default::default() - }, + llm_request: text_request(None, "Analyze a difficult task"), metadata: Some(Metadata { session_id: Some("a".into()), ..Default::default() @@ -120,24 +108,12 @@ fn request() -> Request { } #[tokio::test] -async fn custom_composite_preserves_tier_across_continuations_and_judge_failure() { +async fn custom_composite_reclassifies_user_turns_and_retains_tier_on_failure() { let judge = custom_judge(Some(r#"{"decision":{"target":"capable"}}"#)); let route = custom_route(CONFIG, judge.clone()); let mut request = request(); let first = route.decide(request.clone()).await.unwrap(); assert_eq!(first.selected_model_id().unwrap().as_str(), "strong-model"); - let mut continuation = request.clone(); - continuation - .llm_request - .messages - .push(Message::text(Role::Assistant, "Working")); - let outcome = route.decide(continuation).await.unwrap(); - assert_eq!( - outcome.selected_model_id().unwrap().as_str(), - "strong-model" - ); - assert_eq!(judge.calls.load(Ordering::SeqCst), 1); - for reply in [ None, Some("not JSON"), @@ -163,73 +139,80 @@ async fn custom_composite_preserves_tier_across_continuations_and_judge_failure( request.metadata.as_mut().unwrap().session_id = Some("a".into()); *judge.reply.lock().unwrap() = Some(r#"{"decision":{"target":"efficient"}}"#); let outcome = route.decide(request.clone()).await.unwrap(); - let mut expected = request.llm_request; - expected.model = Some("fast-model".into()); - assert_eq!(outcome.request.llm_request, expected); - assert!(outcome.response.is_none()); + assert_eq!(outcome.selected_model_id().unwrap().as_str(), "fast-model"); } #[tokio::test] -async fn custom_composite_respects_new_session_and_fail_closed() { - let config = CONFIG.replace( - "classify_trigger = \"user_turn\"", - "classify_trigger = \"new_session\"", - ); - let judge = custom_judge(Some(r#"{"decision":{"target":"capable"}}"#)); - let route = custom_route(&config, judge.clone()); - for _ in 0..2 { - let outcome = route.decide(request()).await.unwrap(); - assert_eq!( - outcome.selected_model_id().unwrap().as_str(), - "strong-model" - ); - } - assert_eq!(judge.calls.load(Ordering::SeqCst), 1); - let closed = CONFIG.replace("mode = \"custom\"", "mode = \"custom\"\nfail_open = false"); - let route = custom_route(&closed, custom_judge(None)); - assert!(route.decide(request()).await.is_err()); - - let judge = custom_judge(Some(r#"{"decision":{"target":"capable"}}"#)); - let route = custom_route(&closed, judge.clone()); - route.decide(request()).await.unwrap(); - // Invalid verdicts abstain independently of client-error recovery. - for reply in ["not JSON", "{}", r#"{"decision":{"target":"other"}}"#] { - *judge.reply.lock().unwrap() = Some(reply); - let outcome = route.decide(request()).await.unwrap(); - assert_eq!( - outcome.selected_model_id().unwrap().as_str(), - "strong-model" - ); - assert_eq!( - outcome.metadata.unwrap().evidence.unwrap()["source"], - "retained" - ); +async fn custom_composite_retains_first_verdict_or_fallback() { + for trigger in ["user_turn", "new_session"] { + for hash_keyed in [false, true] { + for reply in [ + Some(r#"{"decision":{"target":"capable"}}"#), + None, + Some("not JSON"), + ] { + let config = CONFIG.replace("user_turn", trigger).replace( + "mode = \"custom\"", + &format!("mode = \"custom\"\nmessage_hash_fallback = {hash_keyed}"), + ); + let judge = custom_judge(reply); + let route = custom_route(&config, judge.clone()); + let mut request = request(); + if hash_keyed { + request.metadata = None; + } + let expected = if reply.is_some_and(|reply| reply.contains("capable")) { + "strong-model" + } else { + "fast-model" + }; + for _ in 0..2 { + let outcome = route.decide(request.clone()).await.unwrap(); + assert_eq!(outcome.selected_model_id().unwrap().as_str(), expected); + request + .llm_request + .messages + .push(Message::text(Role::Assistant, "Working")); + // A recovered judge must not change a new-session assignment. + *judge.reply.lock().unwrap() = Some(r#"{"decision":{"target":"capable"}}"#); + } + assert_eq!( + judge.calls.load(Ordering::SeqCst), + 1, + "{trigger}, hash_keyed={hash_keyed}, reply={reply:?}" + ); + } + } } } #[tokio::test] -async fn custom_composite_retains_tier_with_message_hash_fallback() { - for trigger in ["user_turn", "new_session"] { - let config = CONFIG.replace("user_turn", trigger).replace( - "mode = \"custom\"", - "mode = \"custom\"\nmessage_hash_fallback = true", +async fn custom_composite_fail_closed_rejects_unusable_verdicts() { + let config = CONFIG + .replace("mode = \"custom\"", "mode = \"custom\"\nfail_open = false") + .replace( + r#"["efficient","capable"]"#, + r#"["efficient","capable","judge"]"#, ); - let judge = custom_judge(Some(r#"{"decision":{"target":"capable"}}"#)); - let route = custom_route(&config, judge.clone()); - let mut request = request(); - request.metadata = None; - for _ in 0..2 { - let outcome = route.decide(request.clone()).await.unwrap(); - assert_eq!( - outcome.selected_model_id().unwrap().as_str(), - "strong-model" + for retained in [false, true] { + for reply in [ + None, + Some("not JSON"), + Some("{}"), + Some(r#"{"decision":{"target":"judge"}}"#), + Some(r#"{"decision":{"target":"other"}}"#), + ] { + let judge = custom_judge(Some(r#"{"decision":{"target":"capable"}}"#)); + let route = custom_route(&config, judge.clone()); + if retained { + route.decide(request()).await.unwrap(); + } + *judge.reply.lock().unwrap() = reply; + assert!( + route.decide(request()).await.is_err(), + "retained={retained}, reply={reply:?}" ); - request - .llm_request - .messages - .push(Message::text(Role::Assistant, "Working")); } - assert_eq!(judge.calls.load(Ordering::SeqCst), 1); } } diff --git a/docs/routing_algorithms/composite_routing.md b/docs/routing_algorithms/composite_routing.md index 09d9052aaa..357f68457a 100644 --- a/docs/routing_algorithms/composite_routing.md +++ b/docs/routing_algorithms/composite_routing.md @@ -78,9 +78,11 @@ policy = { type = "target_selector", selector = "/decision/target" } ``` The schema is validated locally. The selector names a tier, and the existing -stage targets supply its model. A missing or invalid verdict retains the previous -tier, or uses efficient for a new session. Client failures do the same with -`fail_open = true` (the default); with `fail_open = false`, client errors propagate. +stage targets supply its model. With `fail_open = true` (the default), a missing +or invalid verdict or a client failure retains the previous tier, or uses efficient +for a new session. A failed `new_session` judgment retains that fallback so tool +continuations do not call the judge again. With `fail_open = false`, client failures +and verdicts that do not select `efficient` or `capable` return an error. Capability thresholds and decision-model settings cannot be combined with `mode = "custom"`. From 115deed02e8529f2585eb149846e47af47268a8b Mon Sep 17 00:00:00 2001 From: ayushag Date: Fri, 16 Oct 2026 10:42:37 -0700 Subject: [PATCH 4/5] fix(composite): preserve configured fallback and focus routing tests Signed-off-by: ayushag --- crates/libsy/src/algorithms/composite.rs | 144 +++++++++- crates/switchyard-runner/src/config.rs | 22 ++ .../tests/custom_composite.rs | 246 ------------------ 3 files changed, 165 insertions(+), 247 deletions(-) delete mode 100644 crates/switchyard-runner/tests/custom_composite.rs diff --git a/crates/libsy/src/algorithms/composite.rs b/crates/libsy/src/algorithms/composite.rs index cb120e05cd..babdaf455f 100644 --- a/crates/libsy/src/algorithms/composite.rs +++ b/crates/libsy/src/algorithms/composite.rs @@ -38,6 +38,7 @@ struct TierSetter { trigger: ClassifyTrigger, message_hash_fallback: bool, require_verdict: bool, + default_tier: Tier, tiers: Mutex>, } @@ -103,7 +104,7 @@ impl Processor for TierSetter { && let Some(identity) = identity.as_ref() { // A failed opening judgment is still the session's one judgment. - self.retain(identity.clone(), Tier::Efficient); + self.retain(identity.clone(), self.default_tier); } } // Either not this request's turn or the judge had no verdict, so the last @@ -191,6 +192,7 @@ impl CompositeRouter { trigger, message_hash_fallback, require_verdict, + default_tier: config.stage.mode.default_tier(), tiers: Mutex::new(HashMap::new()), }; let route = build_stage_route(config.stage)? @@ -311,6 +313,146 @@ mod tests { } } + fn custom_judge(trigger: ClassifyTrigger, fail_open: bool) -> CompositeJudgeConfig { + let mut config = CustomClassifierConfig::new( + "Select a tier", + serde_json::json!({"type": "object"}), + crate::CustomClassifierPolicy::target_selector("/decision/target"), + ); + config.classify_trigger = trigger; + CompositeJudgeConfig::Custom { config, fail_open } + } + + #[tokio::test] + async fn a_failed_session_judgment_retains_the_configured_stage_default() -> Result<()> { + for custom in [false, true] { + for (mode, expected) in [ + (PickerMode::EfficientFirst, "weak"), + (PickerMode::CapableFirst, "strong"), + ] { + let judge = if custom { + custom_judge(ClassifyTrigger::NewSession, true) + } else { + CompositeJudgeConfig::Capability(TaskClassifierConfig { + classify_trigger: ClassifyTrigger::NewSession, + ..Default::default() + }) + }; + let router = Arc::new(CompositeRouter::new(CompositeRouterConfig { + judge, + stage: StageRouterConfig::new(mode, 0.5), + })?); + let calls = Arc::new(Mutex::new(0)); + let serve = { + let calls = calls.clone(); + move |target: ModelId, _request: Request| { + let completion = if target.as_str() == JUDGE { + *calls.lock() += 1; + "not JSON".to_string() + } else { + target.to_string() + }; + async move { Ok(crate::core::testing::reply(completion)) } + } + }; + for request in [ + user_turn_request(), + turn_request(false), + user_turn_request(), + ] { + let (selected, _) = test_drive_with_models( + router.clone(), + request, + runtime_models(), + serve.clone(), + ) + .await?; + assert_eq!(selected.as_str(), expected); + } + assert_eq!( + *calls.lock(), + 1, + "new_session must retain a failed judgment" + ); + } + } + Ok(()) + } + + #[tokio::test] + async fn a_custom_judge_sets_the_tier_and_retains_it_on_failure() -> Result<()> { + let router = Arc::new(CompositeRouter::new(CompositeRouterConfig { + judge: custom_judge(ClassifyTrigger::UserTurn, true), + stage: StageRouterConfig::new(PickerMode::EfficientFirst, 0.5), + })?); + let replies = Arc::new(Mutex::new( + [ + r#"{"decision":{"target":"capable"}}"#, + "not JSON", + r#"{"decision":{"target":"efficient"}}"#, + ] + .into_iter(), + )); + let serve = { + let replies = replies.clone(); + move |target: ModelId, _request: Request| { + let completion = if target.as_str() == JUDGE { + replies + .lock() + .next() + .expect("unexpected judge call") + .to_string() + } else { + target.to_string() + }; + async move { Ok(crate::core::testing::reply(completion)) } + } + }; + for (request, expected) in [ + (user_turn_request(), "strong"), + (turn_request(false), "strong"), + (user_turn_request(), "strong"), + (user_turn_request(), "weak"), + ] { + let (selected, _) = + test_drive_with_models(router.clone(), request, runtime_models(), serve.clone()) + .await?; + assert_eq!(selected.as_str(), expected); + } + assert!(replies.lock().next().is_none()); + Ok(()) + } + + #[tokio::test] + async fn a_custom_judge_failure_stops_composite_when_fail_open_is_false() -> Result<()> { + for reply in [ + None, + Some("not JSON"), + Some(r#"{"decision":{"target":"judge"}}"#), + ] { + let router = Arc::new(CompositeRouter::new(CompositeRouterConfig { + judge: custom_judge(ClassifyTrigger::UserTurn, false), + stage: StageRouterConfig::new(PickerMode::EfficientFirst, 0.5), + })?); + let outcome = test_drive_with_models( + router, + user_turn_request(), + runtime_models(), + move |_target: ModelId, _request: Request| async move { + reply.map(crate::core::testing::reply).ok_or_else(|| { + switchyard_protocol::LlmClientError::General("judge unavailable".into()) + }) + }, + ) + .await; + assert!( + outcome.is_err(), + "unusable verdict must stop routing: {reply:?}" + ); + } + Ok(()) + } + /// A route may point its judge at the same target it serves capable turns on. /// The tier then cannot be recovered by looking the served model up in the /// runtime groups — one id, two categories — so it comes off the verdict itself. diff --git a/crates/switchyard-runner/src/config.rs b/crates/switchyard-runner/src/config.rs index b13b1cfa8b..070afbb286 100644 --- a/crates/switchyard-runner/src/config.rs +++ b/crates/switchyard-runner/src/config.rs @@ -1322,6 +1322,28 @@ evidence = { description = "candidate comparison" } } } + #[test] + fn composite_custom_judge_builds_and_rejects_capability_settings() -> RunnerResult<()> { + let source = composite_config().replace( + "base_threshold = 0.5\nclassify_trigger", + r#"mode = "custom" +prompt = 'Select efficient or capable.' +response_schema = '{"type":"object","properties":{"decision":{"type":"object","properties":{"target":{"type":"string","enum":["efficient","capable"]}},"required":["target"],"additionalProperties":false}},"required":["decision"],"additionalProperties":false}' +policy = { type = "target_selector", selector = "/decision/target" } +classify_trigger"#, + ); + runner_from_toml(&source)?; + let mixed = source.replace( + "mode = \"custom\"", + "mode = \"custom\"\nbase_threshold = 0.5", + ); + assert!( + error_message(&mixed) + .contains("composite custom classifier cannot use capability thresholds") + ); + Ok(()) + } + #[test] fn composite_route_builds_and_claims_both_tiers_and_its_judge() -> RunnerResult<()> { let runner = runner_from_toml(&composite_config())?; diff --git a/crates/switchyard-runner/tests/custom_composite.rs b/crates/switchyard-runner/tests/custom_composite.rs deleted file mode 100644 index c599b00c61..0000000000 --- a/crates/switchyard-runner/tests/custom_composite.rs +++ /dev/null @@ -1,246 +0,0 @@ -// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. -// SPDX-License-Identifier: Apache-2.0 - -use async_trait::async_trait; -use std::collections::BTreeMap; -use std::sync::{ - Arc, Mutex, - atomic::{AtomicUsize, Ordering}, -}; -use switchyard_llm_client::ClientRouter; -use switchyard_protocol::{ - LlmClientError, LlmResponse, Message, Metadata, Request, Response, Role, RoutedLlmClient, - text_request, text_response, -}; -use switchyard_runner::{AlgorithmSpec, ModelCapabilities, Route, Runner}; - -const CONFIG: &str = r#" -schema_version = 1 -[llm_clients.local] -format = "openai_chat" -base_url = "http://127.0.0.1:1/v1" -[targets.fast] -id = "fast-model" -llm_client = "local" -[targets.strong] -id = "strong-model" -llm_client = "local" -[routes.composed] -id = "composite-id" -type = "composite" -[routes.composed.classifier] -target = "fast" -mode = "custom" -prompt = 'Choose efficient or capable for the current task.' -response_schema = '{"type":"object","properties":{"decision":{"type":"object","properties":{"target":{"type":"string","enum":["efficient","capable"]}},"required":["target"],"additionalProperties":false}},"required":["decision"],"additionalProperties":false}' -policy = { type = "target_selector", selector = "/decision/target" } -classify_trigger = "user_turn" -max_output_tokens = 384 -[routes.composed.stage] -efficient_target = "fast" -capable_target = "strong" -"#; - -struct CustomJudge { - reply: Mutex>, - calls: AtomicUsize, -} - -#[async_trait] -impl RoutedLlmClient for CustomJudge { - async fn call(&self, _request: Request) -> Result { - self.calls.fetch_add(1, Ordering::SeqCst); - let reply = self - .reply - .lock() - .unwrap() - .ok_or_else(|| LlmClientError::General("judge unavailable".into()))?; - Ok(Response { - llm_response: LlmResponse::Agg(text_response(None, reply)), - metadata: None, - upstream_headers: Default::default(), - }) - } -} - -fn custom_judge(reply: Option<&'static str>) -> Arc { - Arc::new(CustomJudge { - reply: Mutex::new(reply), - calls: AtomicUsize::new(0), - }) -} - -fn custom_route(config: &str, judge: Arc) -> Route { - let mut config: toml::Value = toml::from_str(config).unwrap(); - config["routes"]["composed"] - .as_table_mut() - .unwrap() - .remove("id"); - let spec: AlgorithmSpec = config["routes"]["composed"].clone().try_into().unwrap(); - let targets = BTreeMap::from([ - ("fast".into(), "fast-model".into()), - ("strong".into(), "strong-model".into()), - ]); - let (algorithm, models) = spec - .build_with_runtime_models("composed", &targets) - .unwrap(); - Route::new( - algorithm, - ClientRouter::single(judge), - None, - ModelCapabilities::default(), - None, - None, - Vec::new(), - models, - ) -} - -fn request() -> Request { - Request { - llm_request: text_request(None, "Analyze a difficult task"), - metadata: Some(Metadata { - session_id: Some("a".into()), - ..Default::default() - }), - ..Default::default() - } -} - -#[tokio::test] -async fn custom_composite_reclassifies_user_turns_and_retains_tier_on_failure() { - let judge = custom_judge(Some(r#"{"decision":{"target":"capable"}}"#)); - let route = custom_route(CONFIG, judge.clone()); - let mut request = request(); - let first = route.decide(request.clone()).await.unwrap(); - assert_eq!(first.selected_model_id().unwrap().as_str(), "strong-model"); - for reply in [ - None, - Some("not JSON"), - Some("{}"), - Some(r#"{"decision":{"target":"judge"}}"#), - Some(r#"{"decision":{"target":"other"}}"#), - ] { - *judge.reply.lock().unwrap() = reply; - let outcome = route.decide(request.clone()).await.unwrap(); - assert_eq!( - outcome.selected_model_id().unwrap().as_str(), - "strong-model" - ); - assert_eq!( - outcome.metadata.unwrap().evidence.unwrap()["source"], - "fail_open" - ); - } - *judge.reply.lock().unwrap() = None; - request.metadata.as_mut().unwrap().session_id = Some("b".into()); - let outcome = route.decide(request.clone()).await.unwrap(); - assert_eq!(outcome.selected_model_id().unwrap().as_str(), "fast-model"); - request.metadata.as_mut().unwrap().session_id = Some("a".into()); - *judge.reply.lock().unwrap() = Some(r#"{"decision":{"target":"efficient"}}"#); - let outcome = route.decide(request.clone()).await.unwrap(); - assert_eq!(outcome.selected_model_id().unwrap().as_str(), "fast-model"); -} - -#[tokio::test] -async fn custom_composite_retains_first_verdict_or_fallback() { - for trigger in ["user_turn", "new_session"] { - for hash_keyed in [false, true] { - for reply in [ - Some(r#"{"decision":{"target":"capable"}}"#), - None, - Some("not JSON"), - ] { - let config = CONFIG.replace("user_turn", trigger).replace( - "mode = \"custom\"", - &format!("mode = \"custom\"\nmessage_hash_fallback = {hash_keyed}"), - ); - let judge = custom_judge(reply); - let route = custom_route(&config, judge.clone()); - let mut request = request(); - if hash_keyed { - request.metadata = None; - } - let expected = if reply.is_some_and(|reply| reply.contains("capable")) { - "strong-model" - } else { - "fast-model" - }; - for _ in 0..2 { - let outcome = route.decide(request.clone()).await.unwrap(); - assert_eq!(outcome.selected_model_id().unwrap().as_str(), expected); - request - .llm_request - .messages - .push(Message::text(Role::Assistant, "Working")); - // A recovered judge must not change a new-session assignment. - *judge.reply.lock().unwrap() = Some(r#"{"decision":{"target":"capable"}}"#); - } - assert_eq!( - judge.calls.load(Ordering::SeqCst), - 1, - "{trigger}, hash_keyed={hash_keyed}, reply={reply:?}" - ); - } - } - } -} - -#[tokio::test] -async fn custom_composite_fail_closed_rejects_unusable_verdicts() { - let config = CONFIG - .replace("mode = \"custom\"", "mode = \"custom\"\nfail_open = false") - .replace( - r#"["efficient","capable"]"#, - r#"["efficient","capable","judge"]"#, - ); - for retained in [false, true] { - for reply in [ - None, - Some("not JSON"), - Some("{}"), - Some(r#"{"decision":{"target":"judge"}}"#), - Some(r#"{"decision":{"target":"other"}}"#), - ] { - let judge = custom_judge(Some(r#"{"decision":{"target":"capable"}}"#)); - let route = custom_route(&config, judge.clone()); - if retained { - route.decide(request()).await.unwrap(); - } - *judge.reply.lock().unwrap() = reply; - assert!( - route.decide(request()).await.is_err(), - "retained={retained}, reply={reply:?}" - ); - } - } -} - -#[test] -fn custom_composite_rejects_incompatible_or_missing_settings() { - let config = CONFIG; - assert!(Runner::from_toml(config).is_ok()); - for invalid in [ - config.replace( - "mode = \"custom\"", - "mode = \"custom\"\nbase_threshold = 0.5", - ), - config.replace("mode = \"custom\"", "mode = \"escalation\""), - config.replace("mode = \"custom\"", "mode = \"capability\""), - config.replace( - "prompt = 'Choose efficient or capable for the current task.'", - "", - ), - config.replace( - "selector = \"/decision/target\"", - "selector = \"/decision/target~2\"", - ), - config.replace( - "classify_trigger = \"user_turn\"", - "classify_trigger = \"every_request\"", - ), - config.replace("response_schema = '", "response_schema = 'not-json"), - ] { - assert!(Runner::from_toml(&invalid).is_err()); - } -} From db428ef0c136ea87240ef6f1de1d84b8bcf2be55 Mon Sep 17 00:00:00 2001 From: ayushag Date: Fri, 16 Oct 2026 11:48:36 -0700 Subject: [PATCH 5/5] refactor(composite): always fall open for custom judges Signed-off-by: ayushag --- crates/libsy/src/algorithms/composite.rs | 100 +++++-------------- crates/switchyard-runner/src/algorithm.rs | 10 +- crates/switchyard-runner/src/config.rs | 27 +++-- docs/routing_algorithms/composite_routing.md | 14 ++- 4 files changed, 58 insertions(+), 93 deletions(-) diff --git a/crates/libsy/src/algorithms/composite.rs b/crates/libsy/src/algorithms/composite.rs index babdaf455f..a054d7ac53 100644 --- a/crates/libsy/src/algorithms/composite.rs +++ b/crates/libsy/src/algorithms/composite.rs @@ -37,7 +37,6 @@ struct TierSetter { judge: Arc>, trigger: ClassifyTrigger, message_hash_fallback: bool, - require_verdict: bool, default_tier: Tier, tiers: Mutex>, } @@ -95,11 +94,6 @@ impl Processor for TierSetter { } return Ok(()); } - if self.require_verdict { - return Err(LibsyError::AlgorithmError { - message: "composite: custom classifier must select efficient or capable".into(), - }); - } if self.trigger == ClassifyTrigger::NewSession && let Some(identity) = identity.as_ref() { @@ -122,13 +116,8 @@ impl Processor for TierSetter { pub enum CompositeJudgeConfig { /// Selects a tier from a capability verdict. Capability(TaskClassifierConfig), - /// Selects a tier using a custom schema and policy. - Custom { - /// Custom classifier contract and trigger settings. - config: CustomClassifierConfig, - /// Retains the previous tier on failure; otherwise requires a valid tier verdict. - fail_open: bool, - }, + /// Selects a tier using a custom schema and policy, retaining the fallback on failure. + Custom(CustomClassifierConfig), } /// A judge stacked over a stage router. @@ -156,7 +145,7 @@ impl CompositeRouter { CompositeJudgeConfig::Capability(judge) => { (judge.classify_trigger, judge.message_hash_fallback) } - CompositeJudgeConfig::Custom { config, .. } => { + CompositeJudgeConfig::Custom(config) => { (config.classify_trigger, config.message_hash_fallback) } }; @@ -165,13 +154,6 @@ impl CompositeRouter { message: "composite: classify_trigger must be user_turn or new_session".to_string(), }); } - let require_verdict = matches!( - &config.judge, - CompositeJudgeConfig::Custom { - fail_open: false, - .. - } - ); let judge: Arc> = match config.judge { CompositeJudgeConfig::Capability(judge) => { // Composite owns the trigger and retention; standalone affinity never runs. @@ -183,15 +165,14 @@ impl CompositeRouter { }, })?) } - CompositeJudgeConfig::Custom { config, fail_open } => { - LlmTaskClassifier::custom_classifier(&config, fail_open)? + CompositeJudgeConfig::Custom(config) => { + LlmTaskClassifier::custom_classifier(&config, true)? } }; let setter = TierSetter { judge, trigger, message_hash_fallback, - require_verdict, default_tier: config.stage.mode.default_tier(), tiers: Mutex::new(HashMap::new()), }; @@ -223,7 +204,7 @@ mod tests { use std::collections::HashMap; use std::sync::Arc; - use switchyard_protocol::{Category, Message, ModelId, Role}; + use switchyard_protocol::{Category, LlmClientError, Message, ModelId, Role}; use super::*; use crate::algorithms::util::stage::PickerMode; @@ -293,14 +274,11 @@ mod tests { fn rejects_every_request_as_a_trigger() { for judge in [ CompositeJudgeConfig::Capability(TaskClassifierConfig::default()), - CompositeJudgeConfig::Custom { - config: CustomClassifierConfig::new( - "Choose a tier", - serde_json::json!({"type": "object"}), - crate::CustomClassifierPolicy::target_selector("/decision/target"), - ), - fail_open: true, - }, + CompositeJudgeConfig::Custom(CustomClassifierConfig::new( + "Choose a tier", + serde_json::json!({"type": "object"}), + crate::CustomClassifierPolicy::target_selector("/decision/target"), + )), ] { let error = CompositeRouter::new(CompositeRouterConfig { judge, @@ -313,14 +291,14 @@ mod tests { } } - fn custom_judge(trigger: ClassifyTrigger, fail_open: bool) -> CompositeJudgeConfig { + fn custom_judge(trigger: ClassifyTrigger) -> CompositeJudgeConfig { let mut config = CustomClassifierConfig::new( "Select a tier", serde_json::json!({"type": "object"}), crate::CustomClassifierPolicy::target_selector("/decision/target"), ); config.classify_trigger = trigger; - CompositeJudgeConfig::Custom { config, fail_open } + CompositeJudgeConfig::Custom(config) } #[tokio::test] @@ -331,7 +309,7 @@ mod tests { (PickerMode::CapableFirst, "strong"), ] { let judge = if custom { - custom_judge(ClassifyTrigger::NewSession, true) + custom_judge(ClassifyTrigger::NewSession) } else { CompositeJudgeConfig::Capability(TaskClassifierConfig { classify_trigger: ClassifyTrigger::NewSession, @@ -382,14 +360,15 @@ mod tests { #[tokio::test] async fn a_custom_judge_sets_the_tier_and_retains_it_on_failure() -> Result<()> { let router = Arc::new(CompositeRouter::new(CompositeRouterConfig { - judge: custom_judge(ClassifyTrigger::UserTurn, true), + judge: custom_judge(ClassifyTrigger::UserTurn), stage: StageRouterConfig::new(PickerMode::EfficientFirst, 0.5), })?); let replies = Arc::new(Mutex::new( [ - r#"{"decision":{"target":"capable"}}"#, - "not JSON", - r#"{"decision":{"target":"efficient"}}"#, + Some(r#"{"decision":{"target":"capable"}}"#), + None, + Some("not JSON"), + Some(r#"{"decision":{"target":"efficient"}}"#), ] .into_iter(), )); @@ -401,17 +380,22 @@ mod tests { .lock() .next() .expect("unexpected judge call") - .to_string() + .map(str::to_string) } else { - target.to_string() + Some(target.to_string()) }; - async move { Ok(crate::core::testing::reply(completion)) } + async move { + completion + .map(crate::core::testing::reply) + .ok_or_else(|| LlmClientError::General("judge unavailable".into())) + } } }; for (request, expected) in [ (user_turn_request(), "strong"), (turn_request(false), "strong"), (user_turn_request(), "strong"), + (user_turn_request(), "strong"), (user_turn_request(), "weak"), ] { let (selected, _) = @@ -423,36 +407,6 @@ mod tests { Ok(()) } - #[tokio::test] - async fn a_custom_judge_failure_stops_composite_when_fail_open_is_false() -> Result<()> { - for reply in [ - None, - Some("not JSON"), - Some(r#"{"decision":{"target":"judge"}}"#), - ] { - let router = Arc::new(CompositeRouter::new(CompositeRouterConfig { - judge: custom_judge(ClassifyTrigger::UserTurn, false), - stage: StageRouterConfig::new(PickerMode::EfficientFirst, 0.5), - })?); - let outcome = test_drive_with_models( - router, - user_turn_request(), - runtime_models(), - move |_target: ModelId, _request: Request| async move { - reply.map(crate::core::testing::reply).ok_or_else(|| { - switchyard_protocol::LlmClientError::General("judge unavailable".into()) - }) - }, - ) - .await; - assert!( - outcome.is_err(), - "unusable verdict must stop routing: {reply:?}" - ); - } - Ok(()) - } - /// A route may point its judge at the same target it serves capable turns on. /// The tier then cannot be recovered by looking the served model up in the /// runtime groups — one id, two categories — so it comes off the verdict itself. diff --git a/crates/switchyard-runner/src/algorithm.rs b/crates/switchyard-runner/src/algorithm.rs index 2bb1964d18..edf4cc287f 100644 --- a/crates/switchyard-runner/src/algorithm.rs +++ b/crates/switchyard-runner/src/algorithm.rs @@ -691,6 +691,11 @@ impl StageClassifierConfig { &self, route_name: &str, ) -> AlgorithmResult { + if self.fail_open == Some(false) { + return Err(AlgorithmConfigError::new( + "composite custom classifiers always fail open; remove fail_open = false", + )); + } if self.decision.is_some() || self.base_threshold.is_some() || self.threshold_step.is_some() @@ -1652,10 +1657,7 @@ fn build_algorithm( stage_config.tool_semantics = stage.tool_semantics.clone(); stage_config.handoff_notes = stage.handoff_notes.clone(); let judge = if matches!(classifier.mode, Some(ClassifierMode::Custom)) { - CompositeJudgeConfig::Custom { - config: classifier.custom_classifier_config(route_name)?, - fail_open: classifier.fail_open.unwrap_or(true), - } + CompositeJudgeConfig::Custom(classifier.custom_classifier_config(route_name)?) } else { CompositeJudgeConfig::Capability( classifier.task_classifier_config(route_name, stage, targets)?, diff --git a/crates/switchyard-runner/src/config.rs b/crates/switchyard-runner/src/config.rs index 070afbb286..288e4586fc 100644 --- a/crates/switchyard-runner/src/config.rs +++ b/crates/switchyard-runner/src/config.rs @@ -1333,14 +1333,25 @@ policy = { type = "target_selector", selector = "/decision/target" } classify_trigger"#, ); runner_from_toml(&source)?; - let mixed = source.replace( - "mode = \"custom\"", - "mode = \"custom\"\nbase_threshold = 0.5", - ); - assert!( - error_message(&mixed) - .contains("composite custom classifier cannot use capability thresholds") - ); + runner_from_toml( + &source.replace("mode = \"custom\"", "mode = \"custom\"\nfail_open = true"), + )?; + for (setting, expected) in [ + ( + "base_threshold = 0.5", + "composite custom classifier cannot use capability thresholds", + ), + ( + "fail_open = false", + "composite custom classifiers always fail open", + ), + ] { + let invalid = source.replace( + "mode = \"custom\"", + &format!("mode = \"custom\"\n{setting}"), + ); + assert!(error_message(&invalid).contains(expected)); + } Ok(()) } diff --git a/docs/routing_algorithms/composite_routing.md b/docs/routing_algorithms/composite_routing.md index 357f68457a..c13e7e47fc 100644 --- a/docs/routing_algorithms/composite_routing.md +++ b/docs/routing_algorithms/composite_routing.md @@ -69,7 +69,6 @@ mode = "custom" target = "judge" classify_trigger = "user_turn" max_output_tokens = 64 -fail_open = true prompt = 'Choose efficient unless capable offers a concrete advantage for this task. Return JSON with decision.target set to efficient or capable.' response_schema = ''' {"type":"object","properties":{"decision":{"type":"object","properties":{"target":{"type":"string","enum":["efficient","capable"]}},"required":["target"],"additionalProperties":false}},"required":["decision"],"additionalProperties":false} @@ -78,13 +77,12 @@ policy = { type = "target_selector", selector = "/decision/target" } ``` The schema is validated locally. The selector names a tier, and the existing -stage targets supply its model. With `fail_open = true` (the default), a missing -or invalid verdict or a client failure retains the previous tier, or uses efficient -for a new session. A failed `new_session` judgment retains that fallback so tool -continuations do not call the judge again. With `fail_open = false`, client failures -and verdicts that do not select `efficient` or `capable` return an error. -Capability thresholds and decision-model -settings cannot be combined with `mode = "custom"`. +stage targets supply its model. Custom classifiers always fall open: a missing +or invalid verdict or a client failure retains the previous tier, or uses the +configured stage default for a new session. A failed `new_session` judgment retains +that fallback so tool continuations do not call the judge again. +`fail_open = false` is rejected for custom classifiers. Capability thresholds and +decision-model settings cannot be combined with `mode = "custom"`. ## Decision model classifier