diff --git a/crates/libsy-llm-client/tests/observability.rs b/crates/libsy-llm-client/tests/observability.rs index 30366dbb2..d014e006c 100644 --- a/crates/libsy-llm-client/tests/observability.rs +++ b/crates/libsy-llm-client/tests/observability.rs @@ -774,6 +774,7 @@ async fn stateful_escalation_warns_once_without_a_session_id() -> switchyard_lib contract: ClassifierContractConfig::default(), config: EscalationJudgeConfig::default(), max_output_tokens: 64, + judge_deadline_ms: None, })?) as Arc; let client = Arc::new(JudgeClient { judge_model: "warning-judge".into(), @@ -824,6 +825,7 @@ async fn deescalation_evidence_stays_pending_until_confirmed() -> switchyard_lib ..EscalationJudgeConfig::default() }, max_output_tokens: 64, + judge_deadline_ms: None, })?) as Arc; let client = |verdict| { Arc::new(JudgeClient { diff --git a/crates/libsy/src/algorithms/escalation.rs b/crates/libsy/src/algorithms/escalation.rs index 7e3d15b83..66c1f2351 100644 --- a/crates/libsy/src/algorithms/escalation.rs +++ b/crates/libsy/src/algorithms/escalation.rs @@ -89,6 +89,7 @@ pub(super) fn build_classifier( contract_config: ClassifierContractConfig, config: EscalationJudgeConfig, max_output_tokens: u64, + judge_deadline_ms: Option, ) -> Result>> { let confirmations = config.confirmations; let deescalation = match config.deescalation { @@ -99,6 +100,7 @@ pub(super) fn build_classifier( config.clone(), Some(EvaluationPhase::Strong), max_output_tokens, + judge_deadline_ms, )?, }), None => None, @@ -110,6 +112,7 @@ pub(super) fn build_classifier( config, is_phase_aware.then_some(EvaluationPhase::Efficient), max_output_tokens, + judge_deadline_ms, )?, confirmations, deescalation, @@ -361,7 +364,7 @@ impl Classifier for EscalationClassifier { let verdict = self .escalation_judge .verdict(state, &judge_request, driver, judge_models) - .await; + .await?; let held = count(state, STREAK_KEY); let held_category = category(state).map(str::to_string); @@ -539,6 +542,7 @@ mod tests { ..EscalationJudgeConfig::default() }, max_output_tokens: DEFAULT_JUDGE_MAX_OUTPUT_TOKENS, + judge_deadline_ms: None, }, )?)) } @@ -553,6 +557,7 @@ mod tests { ..EscalationJudgeConfig::default() }, max_output_tokens: DEFAULT_JUDGE_MAX_OUTPUT_TOKENS, + judge_deadline_ms: None, }, )?)) } @@ -681,6 +686,7 @@ mod tests { ..EscalationJudgeConfig::default() }, max_output_tokens: DEFAULT_JUDGE_MAX_OUTPUT_TOKENS, + judge_deadline_ms: None, })?); test_drive_with_models(router, classify_request(), runtime_models(), serve).await?; diff --git a/crates/libsy/src/algorithms/llm_class.rs b/crates/libsy/src/algorithms/llm_class.rs index e45d73ff7..3469b6984 100644 --- a/crates/libsy/src/algorithms/llm_class.rs +++ b/crates/libsy/src/algorithms/llm_class.rs @@ -352,6 +352,10 @@ pub struct LlmCapabilityConfig { pub contract: ClassifierContractConfig, /// Maximum completion tokens available to the classifier verdict. pub max_output_tokens: u64, + /// Whole-consultation bound on the judge call, in milliseconds. Covers the model + /// call and the response drain; on expiry the judge is treated as unavailable and + /// follows the route's `fail_open` setting. `None` is unbounded. + pub judge_deadline_ms: Option, } impl Default for LlmCapabilityConfig { @@ -361,6 +365,7 @@ impl Default for LlmCapabilityConfig { threshold_step: 0.0, contract: ClassifierContractConfig::default(), max_output_tokens: DEFAULT_JUDGE_MAX_OUTPUT_TOKENS, + judge_deadline_ms: None, } } } @@ -386,6 +391,8 @@ struct TaskClassifierConfigWire { response_format_type: ClassifierResponseFormat, #[serde(default = "default_judge_max_output_tokens")] max_output_tokens: u64, + #[serde(default)] + judge_deadline_ms: Option, } impl<'de> Deserialize<'de> for TaskClassifierConfig { @@ -405,6 +412,7 @@ impl<'de> Deserialize<'de> for TaskClassifierConfig { threshold_step: wire.threshold_step, contract, max_output_tokens: wire.max_output_tokens, + judge_deadline_ms: wire.judge_deadline_ms, }), fail_open: wire.fail_open, classify_trigger: wire.classify_trigger, @@ -523,6 +531,8 @@ pub struct CustomClassifierConfig { pub recent_turn_window: Option, /// Maximum completion tokens available to the classifier verdict. pub max_output_tokens: u64, + /// Whole-consultation bound on the judge call, in milliseconds. `None` is unbounded. + pub judge_deadline_ms: Option, } impl CustomClassifierConfig { @@ -540,6 +550,7 @@ impl CustomClassifierConfig { message_hash_fallback: false, recent_turn_window: None, max_output_tokens: DEFAULT_JUDGE_MAX_OUTPUT_TOKENS, + judge_deadline_ms: None, } } @@ -652,6 +663,8 @@ pub enum LlmClassifierConfig { config: EscalationJudgeConfig, /// Maximum completion tokens available to the escalation verdict. max_output_tokens: u64, + /// Whole-consultation bound on the escalation judge call, in milliseconds. + judge_deadline_ms: Option, }, /// Routes among model categories using a user-supplied schema and policy. Custom { @@ -676,7 +689,8 @@ impl LlmTaskClassifier { contract, config, max_output_tokens, - } => Self::build_escalation(contract, config, max_output_tokens), + judge_deadline_ms, + } => Self::build_escalation(contract, config, max_output_tokens, judge_deadline_ms), LlmClassifierConfig::Custom { default_target, config, @@ -698,7 +712,8 @@ impl LlmTaskClassifier { input, Self::load_capability_contract(&judge.contract)?, SerdeDecoder::new(), - JudgeRuntimeConfig::new(judge.max_output_tokens)?, + JudgeRuntimeConfig::new(judge.max_output_tokens)? + .with_deadline_ms(judge.judge_deadline_ms)?, ), TaskClassifierPolicy::new(&judge), ) @@ -729,6 +744,7 @@ impl LlmTaskClassifier { message_hash_fallback, recent_turn_window, max_output_tokens, + judge_deadline_ms, } = config; let contract = ClassifierContract::from_inner_schema(&prompt, response_schema)?; let policy = match policy { @@ -741,7 +757,7 @@ impl LlmTaskClassifier { TaskInput { recent_turn_window }, contract, JsonSchemaDecoder::new(), - JudgeRuntimeConfig::new(max_output_tokens)?, + JudgeRuntimeConfig::new(max_output_tokens)?.with_deadline_ms(judge_deadline_ms)?, ), policy, )); @@ -760,8 +776,14 @@ impl LlmTaskClassifier { contract_config: ClassifierContractConfig, config: EscalationJudgeConfig, max_output_tokens: u64, + judge_deadline_ms: Option, ) -> Result { - let inner = escalation::build_classifier(contract_config, config, max_output_tokens)?; + let inner = escalation::build_classifier( + contract_config, + config, + max_output_tokens, + judge_deadline_ms, + )?; Ok(Self { route: FallThrough::::new_with_state() .with_name(ALGORITHM_NAME) @@ -1008,6 +1030,60 @@ mod tests { } } + /// The judge stalls before answering; every other target answers normally. + fn slow_judge(delay: std::time::Duration) -> impl Serve { + move |model: ModelId, request: Request| { + let model = model.to_string(); + let slow = model == "judge"; + async move { + if slow { + tokio::time::sleep(delay).await; + } + Ok(Response { + llm_response: LlmResponse::Agg(text_response( + None, + format!("answer from {model}"), + )), + metadata: request.metadata, + upstream_headers: http::HeaderMap::new(), + }) + } + } + } + + /// The judge accepts the call but its stream never carries a chunk. + fn stalled_judge_stream() -> impl Serve { + use futures::StreamExt; + |model: ModelId, request: Request| async move { + let model = model.to_string(); + let llm_response = if model == "judge" { + LlmResponse::Stream(futures::stream::pending().boxed()) + } else { + LlmResponse::Agg(text_response(None, format!("answer from {model}"))) + }; + Ok(Response { + llm_response, + metadata: request.metadata, + upstream_headers: http::HeaderMap::new(), + }) + } + } + + fn deadline_router(fail_open: bool) -> Result> { + Ok(Arc::new(LlmTaskClassifier::new( + LlmClassifierConfig::Capability { + config: TaskClassifierConfig { + judge: CapabilityJudgeConfig::Llm(LlmCapabilityConfig { + judge_deadline_ms: Some(10), + ..llm_config(TEST_THRESHOLD) + }), + fail_open, + ..TaskClassifierConfig::default() + }, + }, + )?)) + } + fn router() -> Result> { Ok(Arc::new(LlmTaskClassifier::new( LlmClassifierConfig::Capability { @@ -1067,6 +1143,131 @@ mod tests { Ok(()) } + #[tokio::test] + async fn a_judge_past_its_deadline_routes_capable() -> Result<()> { + let (selected_model, response) = test_drive_with_models( + deadline_router(true)?, + classify_request(), + runtime_models(), + slow_judge(std::time::Duration::from_millis(500)), + ) + .await?; + + assert_eq!(selected_model, "capable"); + assert_eq!( + response.llm_response.as_agg().map(completion_text), + Some("answer from capable".to_string()) + ); + Ok(()) + } + + #[tokio::test] + async fn a_stalled_judge_stream_is_cut_by_the_deadline() -> Result<()> { + // The judge returned headers promptly, so only a bound on the whole + // consultation — the drain included — can end this turn. + let (selected_model, _) = test_drive_with_models( + deadline_router(true)?, + classify_request(), + runtime_models(), + stalled_judge_stream(), + ) + .await?; + + assert_eq!(selected_model, "capable"); + Ok(()) + } + + #[tokio::test] + async fn a_judge_past_its_deadline_stops_when_not_failing_open() -> Result<()> { + let outcome = test_drive_with_models( + deadline_router(false)?, + classify_request(), + runtime_models(), + slow_judge(std::time::Duration::from_millis(500)), + ) + .await; + let error = outcome + .err() + .expect("a deadline expiry with fail_open = false must stop the request"); + + assert!( + matches!( + error, + LibsyError::ClientCall { + source: LlmClientError::Timeout { .. }, + .. + } + ), + "expected a client timeout, got {error:?}" + ); + Ok(()) + } + + #[test] + fn a_zero_judge_deadline_is_rejected_in_every_mode() { + let capability = LlmClassifierConfig::Capability { + config: TaskClassifierConfig { + judge: CapabilityJudgeConfig::Llm(LlmCapabilityConfig { + judge_deadline_ms: Some(0), + ..llm_config(TEST_THRESHOLD) + }), + ..TaskClassifierConfig::default() + }, + }; + assert!( + LlmTaskClassifier::new(capability).is_err(), + "capability mode must reject judge_deadline_ms = 0" + ); + + let mut custom = CustomClassifierConfig::new( + "Route by topic.", + serde_json::json!({"type": "object"}), + CustomClassifierPolicy::target_selector("/target"), + ); + custom.judge_deadline_ms = Some(0); + assert!( + LlmTaskClassifier::new(LlmClassifierConfig::Custom { + default_target: Category::Capable, + config: custom, + }) + .is_err(), + "custom mode must reject judge_deadline_ms = 0" + ); + + let escalation = LlmClassifierConfig::Escalation { + contract: ClassifierContractConfig::default(), + config: EscalationJudgeConfig::default(), + max_output_tokens: DEFAULT_JUDGE_MAX_OUTPUT_TOKENS, + judge_deadline_ms: Some(0), + }; + assert!( + LlmTaskClassifier::new(escalation).is_err(), + "escalation mode must reject judge_deadline_ms = 0" + ); + } + + #[test] + fn classifier_config_parses_a_judge_deadline() { + let config: TaskClassifierConfig = serde_json::from_value(serde_json::json!({ + "base_threshold": 0.5, + "judge_deadline_ms": 250, + })) + .expect("a configured judge deadline should parse"); + let CapabilityJudgeConfig::Llm(judge) = config.judge else { + panic!("expected the LLM judge variant"); + }; + assert_eq!(judge.judge_deadline_ms, Some(250)); + + let config: TaskClassifierConfig = serde_json::from_value(serde_json::json!({ + "base_threshold": 0.5, + })) + .expect("an unset judge deadline should parse"); + let CapabilityJudgeConfig::Llm(judge) = config.judge else { + panic!("expected the LLM judge variant"); + }; + assert_eq!(judge.judge_deadline_ms, None); + } + #[tokio::test] async fn classifier_judges_each_request_without_affinity() -> Result<()> { let recorder = Arc::new(Recorder::default()); diff --git a/crates/libsy/src/algorithms/util/escalation.rs b/crates/libsy/src/algorithms/util/escalation.rs index 2160cb7ec..c4e165232 100644 --- a/crates/libsy/src/algorithms/util/escalation.rs +++ b/crates/libsy/src/algorithms/util/escalation.rs @@ -283,6 +283,7 @@ pub(crate) fn build_judge( config: EscalationJudgeConfig, phase: Option, max_output_tokens: u64, + judge_deadline_ms: Option, ) -> Result> { config.validate()?; let contract = build_contract(contract_config, phase.is_some())?; @@ -291,7 +292,7 @@ pub(crate) fn build_judge( EscalationInput { config, phase }, contract, SerdeDecoder::new(), - JudgeRuntimeConfig::new(max_output_tokens)?, + JudgeRuntimeConfig::new(max_output_tokens)?.with_deadline_ms(judge_deadline_ms)?, ), EscalationPolicy { phase }, ) diff --git a/crates/libsy/src/algorithms/util/llm_judge.rs b/crates/libsy/src/algorithms/util/llm_judge.rs index 70ff7ef1e..7064982ce 100644 --- a/crates/libsy/src/algorithms/util/llm_judge.rs +++ b/crates/libsy/src/algorithms/util/llm_judge.rs @@ -8,6 +8,7 @@ //! the route. use std::marker::PhantomData; +use std::time::Duration; use async_trait::async_trait; use serde::de::DeserializeOwned; @@ -106,6 +107,8 @@ impl VerdictDecoder for JsonSchemaDecoder { /// Runtime limits shared by structured classifier judges. pub(crate) struct JudgeRuntimeConfig { max_output_tokens: u64, + /// Whole-consultation bound the route places on its judge. + deadline_ms: Option, } impl JudgeRuntimeConfig { @@ -115,7 +118,25 @@ impl JudgeRuntimeConfig { message: "max_output_tokens must be at least 1".to_string(), }); } - Ok(Self { max_output_tokens }) + Ok(Self { + max_output_tokens, + deadline_ms: None, + }) + } + + /// Bounds the whole judge consultation — the model call and the response drain — not + /// only the HTTP call. A `0` deadline is a configuration error, not "unbounded". + pub(crate) fn with_deadline_ms(mut self, deadline_ms: Option) -> Result { + match deadline_ms { + None => {} + Some(0) => { + return Err(LibsyError::AlgorithmError { + message: "judge_deadline_ms must be at least 1".to_string(), + }); + } + Some(deadline_ms) => self.deadline_ms = Some(deadline_ms), + } + Ok(self) } } @@ -178,6 +199,10 @@ where } } + fn deadline_ms(&self) -> Option { + self.runtime.deadline_ms + } + fn parse(&self, response: &AggLlmResponse) -> Result { self.decoder.decode(response, &self.contract) } @@ -189,6 +214,12 @@ pub trait Judge: Send + Sync { fn build_request(&self, state: &State, request: &Request) -> Request; + /// Whole-consultation bound in milliseconds the classifier enforces around the judge + /// call and its response drain. `None` leaves the consultation unbounded. + fn deadline_ms(&self) -> Option { + None + } + fn parse(&self, response: &AggLlmResponse) -> Result { parse_json_verdict(response) } @@ -263,46 +294,86 @@ where } } - /// Consults the judge, yielding `None` when it is unavailable or unintelligible. + /// Consults the judge, yielding `None` when it is unavailable, unintelligible, or + /// over its route deadline. /// /// Errors delivered by the host and invalid verdicts are logged and folded into `None` /// for the policy's fallback branch. The HTTP driver delivers client errors here when - /// error recovery is enabled; otherwise it stops the run. + /// error recovery is enabled; otherwise it stops the run. A route deadline covers the + /// whole consultation — the model call and the response drain — so a judge that + /// answers headers promptly and then stalls mid-stream cannot hold the turn past + /// `judge_deadline_ms` either. On expiry it follows the same recovery split: folded + /// into `None` when recovering, surfaced as a client timeout when not. pub(crate) async fn verdict( &self, state: &mut State, request: &Request, driver: &Driver, judge_models: &[ModelId], - ) -> Option { - let judge_model = judge_models.first()?.as_str(); + ) -> Result> { + let judge_model = judge_models.first().ok_or(LibsyError::NoTargets)?.as_str(); tracing::info!(target = judge_model, "consulting llm judge"); - let response = driver - .call_model_with_error_recovery( - self.judge.build_request(state, request), - judge_models.to_vec(), - self.recover_errors, - ) - .await - .inspect_err(|error| { - self.report_fail_open(driver, safe_error_summary(error), libsy_error_reason(error)); - }) - .ok()?; - let aggregate = response - .llm_response - .into_agg() - .await - .inspect_err(|error| { - self.report_fail_open(driver, safe_client_error(error), client_error_reason(error)); - }) - .ok()?; - self.judge - .parse(&aggregate) - .inspect_err(|error| { - self.report_fail_open(driver, safe_error_summary(error), "parse_error"); - }) - .ok() + let judge_request = self.judge.build_request(state, request); + let consult = async { + let response = driver + .call_model_with_error_recovery( + judge_request, + judge_models.to_vec(), + self.recover_errors, + ) + .await + .inspect_err(|error| { + self.report_fail_open( + driver, + safe_error_summary(error), + libsy_error_reason(error), + ); + }) + .ok()?; + let aggregate = response + .llm_response + .into_agg() + .await + .inspect_err(|error| { + self.report_fail_open( + driver, + safe_client_error(error), + client_error_reason(error), + ); + }) + .ok()?; + self.judge + .parse(&aggregate) + .inspect_err(|error| { + self.report_fail_open(driver, safe_error_summary(error), "parse_error"); + }) + .ok() + }; + let verdict = match self.judge.deadline_ms() { + None => consult.await, + Some(deadline_ms) => { + match tokio::time::timeout(Duration::from_millis(deadline_ms), consult).await { + Ok(verdict) => verdict, + Err(_) => { + let error = format!( + "judge consultation exceeded judge_deadline_ms of {deadline_ms}ms" + ); + self.report_fail_open(driver, error.clone(), "deadline"); + if !self.recover_errors { + return Err(LibsyError::client_call( + ModelId::from(judge_model), + LlmClientError::Timeout { + source: Box::new(std::io::Error::other(error)), + }, + )); + } + None + } + } + } + }; + Ok(verdict) } } @@ -361,7 +432,7 @@ where message: "no models available for category Judge".to_string(), }); } - let verdict = self.verdict(state, request, driver, judge_models).await; + let verdict = self.verdict(state, request, driver, judge_models).await?; let classification = self.policy.to_classification(verdict.as_ref(), driver)?; if let Some(evidence) = self .evidence @@ -723,4 +794,27 @@ mod tests { } Ok(()) } + + #[test] + fn a_zero_judge_deadline_is_a_configuration_error() -> Result<()> { + assert!( + JudgeRuntimeConfig::new(16)? + .with_deadline_ms(Some(0)) + .is_err(), + "judge_deadline_ms = 0 must be rejected, not read as unbounded" + ); + assert_eq!( + JudgeRuntimeConfig::new(16)? + .with_deadline_ms(None)? + .deadline_ms, + None + ); + assert_eq!( + JudgeRuntimeConfig::new(16)? + .with_deadline_ms(Some(250))? + .deadline_ms, + Some(250) + ); + Ok(()) + } } diff --git a/crates/switchyard-py/src/libsy_bindings.rs b/crates/switchyard-py/src/libsy_bindings.rs index 598c830f2..1e863322e 100644 --- a/crates/switchyard-py/src/libsy_bindings.rs +++ b/crates/switchyard-py/src/libsy_bindings.rs @@ -149,6 +149,7 @@ struct PyEscalationClassifierConfig { contract: ClassifierContractConfig, judge: EscalationJudgeConfig, max_output_tokens: u64, + judge_deadline_ms: Option, } #[pymethods] @@ -161,6 +162,7 @@ impl PyEscalationClassifierConfig { window_message_chars=500, deescalation=None, max_output_tokens=4096, + judge_deadline_ms=None, prompt=None, response_format_type="json_schema" ))] @@ -172,6 +174,7 @@ impl PyEscalationClassifierConfig { window_message_chars: usize, deescalation: Option>, max_output_tokens: u64, + judge_deadline_ms: Option, prompt: Option, response_format_type: &str, ) -> PyResult { @@ -186,6 +189,7 @@ impl PyEscalationClassifierConfig { .transpose()?, }, max_output_tokens, + judge_deadline_ms, }) } } @@ -219,7 +223,8 @@ impl PyCustomClassifierConfig { session_affinity=false, message_hash_fallback=false, recent_turn_window=None, - max_output_tokens=4096 + max_output_tokens=4096, + judge_deadline_ms=None ))] #[allow(clippy::too_many_arguments)] fn new( @@ -230,6 +235,7 @@ impl PyCustomClassifierConfig { message_hash_fallback: bool, recent_turn_window: Option, max_output_tokens: u64, + judge_deadline_ms: Option, ) -> PyResult { // Convert the Python schema into serde JSON and pair it with the target-selector policy; // conversion failures propagate to Python through `PyResult`. @@ -242,6 +248,7 @@ impl PyCustomClassifierConfig { inner.message_hash_fallback = message_hash_fallback; inner.recent_turn_window = recent_turn_window; inner.max_output_tokens = max_output_tokens; + inner.judge_deadline_ms = judge_deadline_ms; Ok(Self { inner }) } } @@ -280,6 +287,7 @@ impl PyLlmClassifierConfig { contract: config.contract.clone(), config: config.judge.clone(), max_output_tokens: config.max_output_tokens, + judge_deadline_ms: config.judge_deadline_ms, }, }) } @@ -338,6 +346,7 @@ impl PyTaskClassifierConfig { message_hash_fallback=false, recent_turn_window=None, max_output_tokens=4096, + judge_deadline_ms=None, prompt=None, response_format_type="json_schema" ))] @@ -349,6 +358,7 @@ impl PyTaskClassifierConfig { message_hash_fallback: bool, recent_turn_window: Option, max_output_tokens: u64, + judge_deadline_ms: Option, prompt: Option, response_format_type: &str, ) -> PyResult { @@ -359,6 +369,7 @@ impl PyTaskClassifierConfig { threshold_step, contract: classifier_contract(prompt, response_format_type)?, max_output_tokens, + judge_deadline_ms, }), fail_open: true, classify_trigger: classify_trigger(session_affinity), diff --git a/crates/switchyard-runner/src/algorithm.rs b/crates/switchyard-runner/src/algorithm.rs index a7e40e751..1474d8391 100644 --- a/crates/switchyard-runner/src/algorithm.rs +++ b/crates/switchyard-runner/src/algorithm.rs @@ -124,6 +124,7 @@ struct EscalationClassifierRouteConfig { prompt: Option, response_format_type: ClassifierResponseFormat, max_output_tokens: u64, + judge_deadline_ms: Option, judge: EscalationJudgeConfig, } @@ -138,6 +139,7 @@ struct CustomClassifierRouteConfig { message_hash_fallback: bool, recent_turn_window: Option, max_output_tokens: u64, + judge_deadline_ms: Option, } /// Runtime model groups for a custom classifier, keyed by group name. @@ -275,6 +277,10 @@ pub struct LlmClassifierRouteConfig { /// Most completion tokens the judge verdict may use. #[serde(default = "default_classifier_max_output_tokens")] pub max_output_tokens: u64, + /// Whole-consultation bound on the judge call in milliseconds; expiry follows the + /// route's `fail_open` setting. Unset leaves the consultation unbounded. + #[serde(default)] + pub judge_deadline_ms: Option, /// Escalation mode: how many escalate verdicts latch the session, and how /// much of the transcript the judge sees. pub escalation: Option, @@ -514,6 +520,10 @@ pub struct StageClassifierConfig { /// Most completion tokens the judge verdict may use. #[serde(default = "default_classifier_max_output_tokens")] pub max_output_tokens: u64, + /// Whole-consultation bound on the judge call in milliseconds; expiry follows the + /// route's `fail_open` setting. Unset leaves the consultation unbounded. + #[serde(default)] + pub judge_deadline_ms: Option, } /// The tier pair and scoring settings shared by every stage-router-backed route. @@ -552,6 +562,7 @@ impl StageClassifierConfig { contract: classifier_contract(self.prompt.as_deref()) .with_response_format_type(self.response_format_type), max_output_tokens: self.max_output_tokens, + judge_deadline_ms: self.judge_deadline_ms, }), fail_open: true, classify_trigger: self.classify_trigger, @@ -922,6 +933,7 @@ impl LlmClassifierRouteConfig { prompt, response_format_type, max_output_tokens, + judge_deadline_ms, escalation, decision, models, @@ -971,6 +983,7 @@ impl LlmClassifierRouteConfig { || prompt.is_some() || *response_format_type != ClassifierResponseFormat::JsonSchema || *max_output_tokens != default_classifier_max_output_tokens() + || judge_deadline_ms.is_some() { return Err(AlgorithmConfigError::new(format!( "llm_classifier route {route_name}: decision cannot use LLM judge settings; use decision.cutoff and decision.instructions" @@ -988,6 +1001,7 @@ impl LlmClassifierRouteConfig { contract: classifier_contract(prompt.as_deref()) .with_response_format_type(*response_format_type), max_output_tokens: *max_output_tokens, + judge_deadline_ms: *judge_deadline_ms, }) }; Ok(LlmClassifierModeConfig::Capability( @@ -1051,6 +1065,7 @@ impl LlmClassifierRouteConfig { prompt: prompt.clone(), response_format_type: *response_format_type, max_output_tokens: *max_output_tokens, + judge_deadline_ms: *judge_deadline_ms, judge: required_classifier_field(route_name, "escalation", escalation)?, }, )) @@ -1101,6 +1116,7 @@ impl LlmClassifierRouteConfig { message_hash_fallback: *message_hash_fallback, recent_turn_window: *recent_turn_window, max_output_tokens: *max_output_tokens, + judge_deadline_ms: *judge_deadline_ms, }, )) } @@ -1187,6 +1203,7 @@ fn build_subagent_router_config( ); classifier_config.recent_turn_window = config.recent_turn_window; classifier_config.max_output_tokens = config.max_output_tokens; + classifier_config.judge_deadline_ms = config.judge_deadline_ms; let classifier = Arc::new( LlmTaskClassifier::new(LlmClassifierConfig::Custom { default_target: config.default_target.clone(), @@ -1340,6 +1357,7 @@ fn build_algorithm( .with_response_format_type(config.response_format_type), config: config.judge, max_output_tokens: config.max_output_tokens, + judge_deadline_ms: config.judge_deadline_ms, }) } LlmClassifierModeConfig::Custom(config) => { @@ -1363,6 +1381,7 @@ fn build_algorithm( classifier_config.message_hash_fallback = config.message_hash_fallback; classifier_config.recent_turn_window = config.recent_turn_window; classifier_config.max_output_tokens = config.max_output_tokens; + classifier_config.judge_deadline_ms = config.judge_deadline_ms; LlmTaskClassifier::new(LlmClassifierConfig::Custom { default_target: config.default_target, config: classifier_config, diff --git a/docs/reference/toml_schema.md b/docs/reference/toml_schema.md index ce95015ee..bde115166 100644 --- a/docs/reference/toml_schema.md +++ b/docs/reference/toml_schema.md @@ -267,13 +267,14 @@ checkpoint = "/models/router.pt" ### `llm_classifier` Runs one of three judge-backed modes: `capability`, `escalation`, or `custom`. -`max_output_tokens` applies to all three. +`max_output_tokens` and `judge_deadline_ms` apply to all three. | Key | Required | Default | Meaning | |---|:---:|---|---| | `mode` | No | `capability` | Classifier behavior. Set it explicitly for new configurations. | | `classifier_target` | Capability, escalation | — | Target the judge is called through. Not a routing destination. Custom mode uses `models.judge`. | | `max_output_tokens` | No | `4096` | Maximum completion tokens for the judge verdict. Must be at least `1`. | +| `judge_deadline_ms` | No | unset | Bounds the whole judge consultation in milliseconds, the model call and the response drain included. On expiry the judge is treated as unavailable and follows the route's `fail_open` setting. Must be at least `1`. Unset leaves the consultation unbounded. | | `response_format_type` | No | `json_schema` | Structured-output mode for capability and escalation judges. Use `json_object` when the provider does not support JSON Schema; Switchyard adds the schema to the prompt and validates the verdict locally. Custom mode always uses its configured JSON Schema. | Capability mode classifies before serving. See @@ -398,6 +399,7 @@ configuration. Today a classifier sets the tier a stage router falls open to whe | `classifier.prompt` | No | packaged prompt | Replaces the classifier judge prompt. The verdict schema is unchanged. | | `classifier.response_format_type` | No | `json_schema` | Structured-output mode for the judge. Use `json_object` when the provider does not support JSON Schema. Switchyard adds the schema to the prompt and validates the verdict locally. | | `classifier.max_output_tokens` | No | `4096` | Maximum completion tokens for the judge verdict. Must be at least `1`. | +| `classifier.judge_deadline_ms` | No | unset | Bounds the whole judge consultation in milliseconds, the model call and the response drain included. On expiry the judge is treated as unavailable and the route falls back to the capable tier. Must be at least `1`. Unset leaves the consultation unbounded. | | `stage.capable_target` | Yes | — | Capable tier. | | `stage.efficient_target` | Yes | — | Efficient tier. | | `stage.confidence_threshold` | Yes | — | Corroboration a decisive signal needs. In `[0, 1]`. | diff --git a/docs/routing_algorithms/escalation_router_routing.md b/docs/routing_algorithms/escalation_router_routing.md index 994359a08..089cfe370 100644 --- a/docs/routing_algorithms/escalation_router_routing.md +++ b/docs/routing_algorithms/escalation_router_routing.md @@ -186,7 +186,9 @@ next request. Anchor and transcript caps remain fixed. Set the route-level `max_output_tokens` key to change the judge's reply budget. Any decline or -verdict without new evidence resets the streak to zero. +verdict without new evidence resets the streak to zero. Set `judge_deadline_ms` +to bound the whole judge consultation in milliseconds; on expiry the judge is +treated as unavailable and the session stays where it is. ## Run the route diff --git a/docs/routing_algorithms/llm_classifier_routing.md b/docs/routing_algorithms/llm_classifier_routing.md index de35015ff..77dc2e26e 100644 --- a/docs/routing_algorithms/llm_classifier_routing.md +++ b/docs/routing_algorithms/llm_classifier_routing.md @@ -125,6 +125,7 @@ for the server merge behavior. | `prompt` | packaged capability prompt | Replaces the classifier's system prompt. The packaged verdict schema and routing policy remain active. | | `response_format_type` | `json_schema` | Structured-output mode for capability and escalation judges. Use `json_object` for providers without JSON Schema support. | | `max_output_tokens` | `4096` | Maximum completion tokens available to the classifier verdict. Must be at least `1`. | +| `judge_deadline_ms` | unset | Bounds the whole judge consultation in milliseconds, the model call and the response drain included. On expiry the judge is treated as unavailable and follows `fail_open`: the route either falls back to the capable tier (default) or stops with a 504. Must be at least `1`. Unset leaves the consultation unbounded. | ### Override the classifier prompt