diff --git a/benchmark/server-configs/tb-lite-llm-classifier-opus-kimi-gemini.toml b/benchmark/server-configs/tb-lite-llm-classifier-opus-kimi-gemini.toml index d815dc877..7680386fa 100644 --- a/benchmark/server-configs/tb-lite-llm-classifier-opus-kimi-gemini.toml +++ b/benchmark/server-configs/tb-lite-llm-classifier-opus-kimi-gemini.toml @@ -31,5 +31,5 @@ strong_target = "strong" weak_target = "weak" base_threshold = 0.5 threshold_step = 0.0 -session_affinity = true +classify_trigger = "new_session" message_hash_fallback = true diff --git a/crates/libsy-llm-client/tests/observability.rs b/crates/libsy-llm-client/tests/observability.rs index 60f871783..b90bb8236 100644 --- a/crates/libsy-llm-client/tests/observability.rs +++ b/crates/libsy-llm-client/tests/observability.rs @@ -34,9 +34,9 @@ use tracing_subscriber::layer::{Context as LayerContext, SubscriberExt}; use tracing_subscriber::registry::LookupSpan; use switchyard_libsy::{ - AffinityRouter, Algorithm, Classifier, Driver, LibsyError, LlmClassifierConfig, - LlmTaskClassifier, PickerMode, RoutingOutcome, StageRouter, StageRouterConfig, Step, - TaskClassifierConfig, + AffinityRouter, Algorithm, Classifier, ClassifyTrigger, Driver, LibsyError, + LlmClassifierConfig, LlmTaskClassifier, PickerMode, RoutingOutcome, StageRouter, + StageRouterConfig, Step, TaskClassifierConfig, }; use switchyard_llm_client::{ClientRouter, RunObservation, RunObserver}; use switchyard_protocol::ModelId; @@ -676,7 +676,7 @@ async fn affinity_keeps_the_algorithm_selection_after_client_fallback() capable_target: "affinity-fallback-strong".into(), config: TaskClassifierConfig { base_threshold: 0.5, - session_affinity: true, + classify_trigger: ClassifyTrigger::NewSession, ..TaskClassifierConfig::default() }, })?) as Arc; diff --git a/crates/libsy/src/algorithms/llm_class.rs b/crates/libsy/src/algorithms/llm_class.rs index 589a20158..86c22bb54 100644 --- a/crates/libsy/src/algorithms/llm_class.rs +++ b/crates/libsy/src/algorithms/llm_class.rs @@ -12,7 +12,6 @@ use serde_json::Value; use switchyard_protocol::{ContentBlock, Message, ModelId, Role}; use super::fall_through::{DefaultTarget, FallThrough}; -use super::util::DEFAULT_JUDGE_MAX_OUTPUT_TOKENS; use super::util::affinity::AffinityRouter; use super::util::classifier_contract::{ ClassifierContract, ClassifierContractConfig, ClassifierResponseFormat, @@ -23,6 +22,8 @@ use super::util::llm_judge::{ SerdeDecoder, StructuredJudge, }; use super::util::target_selector::TargetSelectorPolicy; +use super::util::turn_pin::{ClassifyTrigger, TurnPin}; +use super::util::{DEFAULT_JUDGE_MAX_OUTPUT_TOKENS, decisive}; use crate::core::algorithm::{self, Algorithm, Driver}; use crate::core::classifier::{Classification, Classifier, Score}; use crate::core::state::{State, StateValue}; @@ -231,8 +232,8 @@ pub struct TaskClassifierConfig { /// Supported verdicts use `base_threshold`, uncertain and unmatched verdicts use one /// step, and unsupported verdicts use two steps. pub threshold_step: f64, - /// Enables session affinity before the judge-backed classifier. - pub session_affinity: bool, + /// How often the classifier re-decides this session's target. + pub classify_trigger: ClassifyTrigger, /// Uses the first user message as the SessionKey for sticky routing when session metadata is unavailable. pub message_hash_fallback: bool, /// Trailing conversation turns the judge sees on top of the client @@ -256,7 +257,7 @@ struct TaskClassifierConfigWire { #[serde(default)] threshold_step: f64, #[serde(default)] - session_affinity: bool, + classify_trigger: ClassifyTrigger, #[serde(default)] message_hash_fallback: bool, #[serde(default)] @@ -283,7 +284,7 @@ impl<'de> Deserialize<'de> for TaskClassifierConfig { Ok(Self { base_threshold: wire.base_threshold, threshold_step: wire.threshold_step, - session_affinity: wire.session_affinity, + classify_trigger: wire.classify_trigger, message_hash_fallback: wire.message_hash_fallback, recent_turn_window: wire.recent_turn_window, contract, @@ -301,7 +302,7 @@ impl Default for TaskClassifierConfig { Self { base_threshold: 0.0, threshold_step: 0.0, - session_affinity: false, + classify_trigger: ClassifyTrigger::default(), message_hash_fallback: false, recent_turn_window: None, contract: ClassifierContractConfig::default(), @@ -342,9 +343,10 @@ impl TaskClassifierConfig { message: "max_output_tokens must be at least 1".to_string(), }); } - if self.message_hash_fallback && !self.session_affinity { + if self.message_hash_fallback && self.classify_trigger != ClassifyTrigger::NewSession { return Err(LibsyError::AlgorithmError { - message: "message_hash_fallback requires session_affinity".to_string(), + message: "message_hash_fallback requires classify_trigger = new_session" + .to_string(), }); } Ok(()) @@ -379,8 +381,8 @@ pub struct CustomClassifierConfig { pub response_schema: Value, /// Deterministic policy applied after the verdict passes schema validation. pub policy: CustomClassifierPolicy, - /// Enables session affinity before the judge-backed classifier. - pub session_affinity: bool, + /// How often the classifier re-decides this session's target. + pub classify_trigger: ClassifyTrigger, /// Uses the first user message when session metadata is unavailable. pub message_hash_fallback: bool, /// Trailing conversation turns shown to the classifier judge. @@ -400,7 +402,7 @@ impl CustomClassifierConfig { prompt: prompt.into(), response_schema, policy, - session_affinity: false, + classify_trigger: ClassifyTrigger::default(), message_hash_fallback: false, recent_turn_window: None, max_output_tokens: DEFAULT_JUDGE_MAX_OUTPUT_TOKENS, @@ -413,9 +415,10 @@ impl CustomClassifierConfig { message: "max_output_tokens must be at least 1".to_string(), }); } - if self.message_hash_fallback && !self.session_affinity { + if self.message_hash_fallback && self.classify_trigger != ClassifyTrigger::NewSession { return Err(LibsyError::AlgorithmError { - message: "message_hash_fallback requires session_affinity".to_string(), + message: "message_hash_fallback requires classify_trigger = new_session" + .to_string(), }); } Ok(()) @@ -454,13 +457,6 @@ fn streak(state: &State) -> u32 { } } -fn decisive(target: &ModelId) -> Classification { - Classification::Scores(vec![Score { - target: target.clone(), - confidence: 1.0, - }]) -} - fn assistant_message(response: &AggLlmResponse) -> Message { Message { role: Role::Assistant, @@ -592,7 +588,7 @@ pub struct LlmTaskClassifier { struct ClassifierRouteConfig { default_target: ModelId, - session_affinity: bool, + classify_trigger: ClassifyTrigger, message_hash_fallback: bool, } @@ -687,7 +683,7 @@ impl LlmTaskClassifier { config.validate()?; let contract = Self::load_capability_contract(&config.contract)?; let targets = vec![efficient_target.clone(), capable_target.clone()]; - let session_affinity = config.session_affinity; + let classify_trigger = config.classify_trigger; let message_hash_fallback = config.message_hash_fallback; let classifier = Arc::new(TaskClassifier { classifier: JudgeClassifier::new( @@ -715,7 +711,7 @@ impl LlmTaskClassifier { inner, ClassifierRouteConfig { default_target: classifier.capable_target.clone(), - session_affinity, + classify_trigger, message_hash_fallback, }, ) @@ -772,7 +768,7 @@ impl LlmTaskClassifier { prompt, response_schema, policy, - session_affinity, + classify_trigger, message_hash_fallback, recent_turn_window, max_output_tokens, @@ -801,7 +797,7 @@ impl LlmTaskClassifier { classifier, ClassifierRouteConfig { default_target: default_name, - session_affinity, + classify_trigger, message_hash_fallback, }, ) @@ -853,16 +849,24 @@ impl LlmTaskClassifier { config: ClassifierRouteConfig, ) -> Result { algorithm::ensure_model_is_target(&targets, &config.default_target)?; - if config.message_hash_fallback && !config.session_affinity { + if config.message_hash_fallback && config.classify_trigger != ClassifyTrigger::NewSession { return Err(LibsyError::AlgorithmError { - message: "message_hash_fallback requires session_affinity".to_string(), + message: "message_hash_fallback requires classify_trigger = new_session" + .to_string(), }); } + // Wraps the classifier rather than the route, so the pin also holds when this is + // embedded in another cascade and only `score` is called. + let inner = if config.classify_trigger == ClassifyTrigger::UserTurn { + Arc::new(TurnPin::new(inner)) as Arc> + } else { + inner + }; // Affinity comes first so a retained assignment short-circuits the judge call. // Note: when this classifier is embedded inside another cascade (e.g. StageRouter) // the affinity processor never fires — only the inner score() is called. let mut route = FallThrough::::new_with_state(targets).with_name(ALGORITHM_NAME); - if config.session_affinity { + if config.classify_trigger == ClassifyTrigger::NewSession { let affinity = if config.message_hash_fallback { AffinityRouter::new().with_message_hash_fallback() } else { @@ -1206,14 +1210,14 @@ mod tests { } #[tokio::test] - async fn classifier_config_enables_session_affinity() -> Result<()> { + async fn classifier_config_enables_new_session_trigger() -> Result<()> { let recorder = Arc::new(Recorder::default()); let router = Arc::new(LlmTaskClassifier::new(LlmClassifierConfig::Capability { judge_target: ModelId::from("judge"), efficient_target: ModelId::from("efficient"), capable_target: ModelId::from("capable"), config: TaskClassifierConfig { - session_affinity: true, + classify_trigger: ClassifyTrigger::NewSession, ..test_config(TEST_THRESHOLD) }, })?); @@ -1234,7 +1238,7 @@ mod tests { efficient_target: ModelId::from("efficient"), capable_target: ModelId::from("capable"), config: TaskClassifierConfig { - session_affinity: true, + classify_trigger: ClassifyTrigger::NewSession, message_hash_fallback: true, recent_turn_window: None, ..test_config(TEST_THRESHOLD) diff --git a/crates/libsy/src/algorithms/stage.rs b/crates/libsy/src/algorithms/stage.rs index 00ec3e59e..289aa08b5 100644 --- a/crates/libsy/src/algorithms/stage.rs +++ b/crates/libsy/src/algorithms/stage.rs @@ -72,7 +72,7 @@ pub struct LlmFallback { pub judge_target: ModelId, /// Judge configuration. `recent_turn_window` is worth setting to this router's /// `recent_window` so the judge reads the same span the signal scorer scored. - /// Note: `session_affinity` and `message_hash_fallback` have no effect here — + /// Note: `classify_trigger = new_session` and `message_hash_fallback` have no effect here — /// the judge runs as a cascade classifier, not a standalone algorithm. pub config: TaskClassifierConfig, } diff --git a/crates/libsy/src/algorithms/util.rs b/crates/libsy/src/algorithms/util.rs index 66b5ed982..0f115ae55 100644 --- a/crates/libsy/src/algorithms/util.rs +++ b/crates/libsy/src/algorithms/util.rs @@ -10,6 +10,19 @@ pub(crate) mod stage; pub mod subagent; pub(crate) mod target_selector; pub(crate) mod tool_signals; +pub mod turn_pin; + +use switchyard_protocol::ModelId; + +use crate::core::classifier::{Classification, Score}; + +/// A single full-confidence recommendation. +pub(crate) fn decisive(target: &ModelId) -> Classification { + Classification::Scores(vec![Score { + target: target.clone(), + confidence: 1.0, + }]) +} /// Default completion budget for internal classifier and escalation judge calls. pub(crate) const DEFAULT_JUDGE_MAX_OUTPUT_TOKENS: u64 = 4_096; diff --git a/crates/libsy/src/algorithms/util/turn_pin.rs b/crates/libsy/src/algorithms/util/turn_pin.rs new file mode 100644 index 000000000..d86250e77 --- /dev/null +++ b/crates/libsy/src/algorithms/util/turn_pin.rs @@ -0,0 +1,288 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +//! Holds one classifier verdict across the tool-call turns that follow a user message. + +use std::sync::Arc; + +use async_trait::async_trait; +use serde::Deserialize; +use switchyard_protocol::{ContentBlock, Message, ModelId, Request, Response, Role}; + +use super::decisive; +use crate::Result; +use crate::core::algorithm::Driver; +use crate::core::classifier::{Classification, Classifier}; +use crate::core::state::{State, StateValue}; + +/// `State.extra` key holding the pinned target. +const PINNED_TARGET_KEY: &str = "classifier_pinned_target"; + +/// How often the classifier re-decides a session's target. +#[derive(Clone, Copy, Debug, Default, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "snake_case")] +pub enum ClassifyTrigger { + /// Judge every request, tool continuations included. + #[default] + EveryRequest, + /// Judge each new user message and hold that target across the tool calls between. + UserTurn, + /// Judge once and reuse that target for the session. + NewSession, +} + +/// True when the user spoke. Anthropic carries tool results as a `Role::User` message, so +/// role alone cannot tell a human turn from a tool continuation. The same invariant is +/// encoded in the Anthropic codec's `message_is_tool_result_only`. +fn is_user_turn(message: &Message) -> bool { + message.role == Role::User + && !message + .content + .iter() + .all(|block| matches!(block, ContentBlock::ToolResult(_))) +} + +/// True when the conversation ends with the user speaking, so the agent is not mid-task. +fn starts_new_turn(messages: &[Message]) -> bool { + messages.last().is_some_and(is_user_turn) +} + +fn pinned_target(state: &State) -> Option { + match state.extra.get(PINNED_TARGET_KEY) { + Some(StateValue::String(target)) => Some(ModelId::new(target.clone())), + _ => None, + } +} + +/// Pins the inner classifier's verdict until the user speaks again. +/// +/// Without a session id there is no retained state, and every turn is classified. +pub(crate) struct TurnPin { + inner: Arc>, +} + +impl TurnPin { + pub(crate) fn new(inner: Arc>) -> Self { + Self { inner } + } +} + +#[async_trait] +impl Classifier for TurnPin { + fn routing_tier(&self, selected_model_id: &ModelId) -> Option<&'static str> { + self.inner.routing_tier(selected_model_id) + } + + async fn score( + &self, + state: &mut State, + request: &mut Request, + driver: Option<&Driver>, + ) -> Result<(Classification, Option)> { + if let Some(target) = pinned_target(state) + && !starts_new_turn(&request.llm_request.messages) + { + return Ok((decisive(&target), None)); + } + + let (classification, response) = self.inner.score(state, request, driver).await?; + // An abstention clears the pin. Holding the old target would serve this turn from the + // fall-through default while its tool continuations reused the previous turn's target. + match classification.argmax(false)? { + Some(score) => { + state.extra.insert( + PINNED_TARGET_KEY.to_string(), + StateValue::String(score.target.as_str().to_string()), + ); + } + None => { + state.extra.remove(PINNED_TARGET_KEY); + } + } + Ok((classification, response)) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + use parking_lot::Mutex; + use serde_json::Value; + use switchyard_protocol::{LlmRequest, ToolCall, ToolResult}; + + /// Returns queued verdicts in order, counting each consultation. + struct RecordingClassifier { + verdicts: Mutex>>, + consultations: Mutex, + } + + impl RecordingClassifier { + fn new(verdicts: Vec>) -> Arc { + Arc::new(Self { + verdicts: Mutex::new(verdicts.into_iter().rev().collect()), + consultations: Mutex::new(0), + }) + } + + fn consultations(&self) -> u32 { + *self.consultations.lock() + } + } + + #[async_trait] + impl Classifier for RecordingClassifier { + async fn score( + &self, + _state: &mut State, + _request: &mut Request, + _driver: Option<&Driver>, + ) -> Result<(Classification, Option)> { + *self.consultations.lock() += 1; + let verdict = self.verdicts.lock().pop().flatten(); + Ok(( + match verdict { + Some(target) => decisive(&ModelId::new(target)), + None => Classification::Scores(Vec::new()), + }, + None, + )) + } + } + + fn user(text: &str) -> Message { + Message::text(Role::User, text) + } + + fn tool_result(id: &str) -> Message { + Message { + role: Role::User, + content: vec![ContentBlock::ToolResult(ToolResult { + tool_call_id: id.to_string(), + content: Vec::new(), + is_error: None, + })], + } + } + + fn tool_call(id: &str) -> Message { + Message { + role: Role::Assistant, + content: vec![ContentBlock::ToolCall(ToolCall { + id: id.to_string(), + name: "read".to_string(), + arguments: Value::Null, + })], + } + } + + async fn selected( + pin: &TurnPin, + state: &mut State, + messages: Vec, + ) -> Result> { + let mut request = Request { + llm_request: LlmRequest { + messages, + ..LlmRequest::default() + }, + raw_request: None, + metadata: None, + }; + let (classification, _) = pin.score(state, &mut request, None).await?; + Ok(classification + .argmax(false)? + .map(|score| score.target.as_str().to_string())) + } + + #[tokio::test] + async fn tool_continuation_turns_hold_the_pinned_target() -> Result<()> { + let inner = RecordingClassifier::new(vec![Some("capable"), Some("efficient")]); + let pin = TurnPin::new(inner.clone()); + let mut state = State::default(); + let opening = vec![user("debug this")]; + + assert_eq!( + selected(&pin, &mut state, opening.clone()).await?, + Some("capable".to_string()) + ); + let mut continued = opening; + continued.push(tool_call("call-1")); + continued.push(tool_result("call-1")); + assert_eq!( + selected(&pin, &mut state, continued).await?, + Some("capable".to_string()) + ); + assert_eq!(inner.consultations(), 1); + Ok(()) + } + + #[tokio::test] + async fn a_fresh_user_message_re_classifies() -> Result<()> { + let inner = RecordingClassifier::new(vec![Some("efficient"), Some("capable")]); + let pin = TurnPin::new(inner.clone()); + let mut state = State::default(); + let opening = vec![user("summarise this file")]; + + assert_eq!( + selected(&pin, &mut state, opening.clone()).await?, + Some("efficient".to_string()) + ); + let mut followed_up = opening; + followed_up.push(tool_call("call-1")); + followed_up.push(tool_result("call-1")); + followed_up.push(user("now find the race condition")); + assert_eq!( + selected(&pin, &mut state, followed_up).await?, + Some("capable".to_string()) + ); + assert_eq!(inner.consultations(), 2); + Ok(()) + } + + #[tokio::test] + async fn an_abstention_clears_an_earlier_pin() -> Result<()> { + let inner = RecordingClassifier::new(vec![Some("efficient"), None, Some("capable")]); + let pin = TurnPin::new(inner.clone()); + let mut state = State::default(); + let opening = vec![user("summarise this file")]; + + assert_eq!( + selected(&pin, &mut state, opening.clone()).await?, + Some("efficient".to_string()) + ); + let mut followed_up = opening; + followed_up.push(user("now find the race condition")); + assert_eq!(selected(&pin, &mut state, followed_up.clone()).await?, None); + + // The abstaining turn is served by the fall-through default, so its tool + // continuations must not reuse the target pinned by the previous turn. + followed_up.push(tool_call("call-1")); + followed_up.push(tool_result("call-1")); + assert_eq!( + selected(&pin, &mut state, followed_up).await?, + Some("capable".to_string()) + ); + assert_eq!(inner.consultations(), 3); + Ok(()) + } + + #[test] + fn a_turn_starts_only_when_the_user_spoke_last() { + let mut messages = vec![user("debug this")]; + assert!(starts_new_turn(&messages)); + messages.push(tool_call("call-1")); + assert!(!starts_new_turn(&messages)); + // Anthropic sends this as a user-role message, so it must not count as a turn. + messages.push(tool_result("call-1")); + assert!(!starts_new_turn(&messages)); + messages.push(user("still broken")); + assert!(starts_new_turn(&messages)); + assert!(!starts_new_turn(&[])); + + // A tool result the user appended to counts, because the user did speak. + let mut mixed = tool_result("call-2"); + mixed.content.extend(user("and also rename foo").content); + assert!(starts_new_turn(&[mixed])); + } +} diff --git a/crates/libsy/src/lib.rs b/crates/libsy/src/lib.rs index 1bb46db73..9abf46da4 100644 --- a/crates/libsy/src/lib.rs +++ b/crates/libsy/src/lib.rs @@ -31,6 +31,7 @@ pub use algorithms::util::escalation::EscalationJudgeConfig; pub use algorithms::util::prompts::{SystemPromptProcessor, TargetPrompts, append_note}; pub use algorithms::util::subagent::SubagentOverride; pub use algorithms::util::tool_signals::{DEFAULT_RECENT_WINDOW, ToolSignals}; +pub use algorithms::util::turn_pin::ClassifyTrigger; // Stage-router scoring and tier selection — the shared signal-driven routing // core (scorer, picker, and the `StageClassifier`). diff --git a/crates/switchyard-py/src/libsy_bindings.rs b/crates/switchyard-py/src/libsy_bindings.rs index 407a2080b..e2d6a4eaf 100644 --- a/crates/switchyard-py/src/libsy_bindings.rs +++ b/crates/switchyard-py/src/libsy_bindings.rs @@ -12,7 +12,7 @@ use pyo3::exceptions::{PyBaseException, PyStopAsyncIteration, PyTypeError, PyVal use pyo3::prelude::*; use serde_json::Value; use switchyard_libsy::{ - Algorithm, CallModel, ClassifierContractConfig, ClassifierResponseFormat, + Algorithm, CallModel, ClassifierContractConfig, ClassifierResponseFormat, ClassifyTrigger, CustomClassifierConfig, CustomClassifierPolicy, EscalationJudgeConfig, HandoffNoteConfig, LibsyError as RustLibsyError, LlmClassifierConfig, LlmFallback, LlmTaskClassifier, Noop, PickerMode, Random, RoutingOutcome, StageRouter, StageRouterConfig, Step as RustStep, @@ -27,6 +27,15 @@ use tokio::sync::Mutex; use crate::errors::{ContextWindowExceededError, py_libsy_error}; use crate::py_serde::{from_python, to_python}; +/// The Python API keeps its `session_affinity` flag, which selects the per-session trigger. +fn classify_trigger(session_affinity: bool) -> ClassifyTrigger { + if session_affinity { + ClassifyTrigger::NewSession + } else { + ClassifyTrigger::EveryRequest + } +} + /// Convert Python-owned headers into the request metadata expected by libsy. fn header_map_from_python(headers: &HashMap) -> PyResult { let mut result = http::HeaderMap::new(); @@ -155,7 +164,7 @@ impl PyCustomClassifierConfig { from_python::(response_schema)?, CustomClassifierPolicy::target_selector(selector), ); - inner.session_affinity = session_affinity; + inner.classify_trigger = classify_trigger(session_affinity); inner.message_hash_fallback = message_hash_fallback; inner.recent_turn_window = recent_turn_window; inner.max_output_tokens = max_output_tokens; @@ -273,7 +282,7 @@ impl PyTaskClassifierConfig { inner: TaskClassifierConfig { base_threshold, threshold_step, - session_affinity, + classify_trigger: classify_trigger(session_affinity), message_hash_fallback, recent_turn_window, contract: classifier_contract(prompt, response_format_type)?, diff --git a/crates/switchyard-server/README.md b/crates/switchyard-server/README.md index 75d1b1fd1..2ff1a7a5a 100644 --- a/crates/switchyard-server/README.md +++ b/crates/switchyard-server/README.md @@ -107,8 +107,8 @@ routes to `weak_target` or `strong_target`. Beyond the three targets it accepts |---|---|---| | `base_threshold` | *required* | Lowest solve probability that routes a task to `weak_target`. Raise it to send less traffic to the weak model. | | `threshold_step` | `0.0` | Finite, non-negative amount added once for uncertain or unmatched verdicts and twice for unsupported verdicts. `base_threshold + 2 * threshold_step` must be at most `1`. | -| `session_affinity` | `false` | Reuses a session's first routing decision on later turns, so the judge is called once per session rather than once per turn. | -| `message_hash_fallback` | `false` | Extends affinity to clients that send no session header, keying on the first user message. Requires `session_affinity = true`. | +| `classify_trigger` | `every_request` | When the judge runs. `every_request` judges every request including tool continuations, `user_turn` judges each new user message and holds that target across the tool calls between, `new_session` judges once and reuses that target for the session. | +| `message_hash_fallback` | `false` | Extends affinity to clients that send no session header, keying on the first user message. Requires `classify_trigger = "new_session"`. | Session affinity retains a decision for the process lifetime, including a `strong_target` fallback produced while the judge was unreachable. `message_hash_fallback` keys on request diff --git a/crates/switchyard-server/src/config.rs b/crates/switchyard-server/src/config.rs index 420ac5e2b..e45652238 100644 --- a/crates/switchyard-server/src/config.rs +++ b/crates/switchyard-server/src/config.rs @@ -10,9 +10,10 @@ use std::sync::Arc; use libsy::{ AdvisorGate, AdvisorGateConfig, Algorithm, ClassifierContractConfig, ClassifierResponseFormat, - CustomClassifierConfig, CustomClassifierPolicy, EscalationJudgeConfig, GateTrigger, - HandoffNoteConfig, LlmClassifierConfig, LlmFallback, LlmTaskClassifier, Noop, Passthrough, - PickerMode, Random, StageRouter, StageRouterConfig, TargetPrompts, TaskClassifierConfig, + ClassifyTrigger, CustomClassifierConfig, CustomClassifierPolicy, EscalationJudgeConfig, + GateTrigger, HandoffNoteConfig, LlmClassifierConfig, LlmFallback, LlmTaskClassifier, Noop, + Passthrough, PickerMode, Random, StageRouter, StageRouterConfig, TargetPrompts, + TaskClassifierConfig, }; use serde::Deserialize; use serde_json::Value; @@ -344,7 +345,7 @@ struct CapabilityClassifierRouteConfig { weak_target: String, base_threshold: f64, threshold_step: f64, - session_affinity: bool, + classify_trigger: ClassifyTrigger, message_hash_fallback: bool, recent_turn_window: Option, prompt: Option, @@ -369,7 +370,7 @@ struct CustomClassifierRouteConfig { prompt: String, response_schema: String, policy: ClassifierPolicyConfig, - session_affinity: bool, + classify_trigger: ClassifyTrigger, message_hash_fallback: bool, recent_turn_window: Option, max_output_tokens: u64, @@ -429,7 +430,7 @@ enum RouteConfig { #[serde(default)] threshold_step: Option, #[serde(default)] - session_affinity: bool, + classify_trigger: ClassifyTrigger, #[serde(default)] message_hash_fallback: bool, #[serde(default)] @@ -538,7 +539,7 @@ struct StageClassifierConfig { #[serde(default)] threshold_step: f64, #[serde(default)] - session_affinity: bool, + classify_trigger: ClassifyTrigger, #[serde(default)] message_hash_fallback: bool, #[serde(default)] @@ -556,7 +557,7 @@ impl StageClassifierConfig { TaskClassifierConfig { base_threshold: self.base_threshold, threshold_step: self.threshold_step, - session_affinity: self.session_affinity, + classify_trigger: self.classify_trigger, message_hash_fallback: self.message_hash_fallback, recent_turn_window: self.recent_turn_window, contract: classifier_contract(self.prompt.as_deref()) @@ -695,7 +696,7 @@ impl RouteConfig { weak_target, base_threshold, threshold_step, - session_affinity, + classify_trigger, message_hash_fallback, recent_turn_window, prompt, @@ -753,7 +754,7 @@ impl RouteConfig { base_threshold, )?, threshold_step: threshold_step.unwrap_or_default(), - session_affinity: *session_affinity, + classify_trigger: *classify_trigger, message_hash_fallback: *message_hash_fallback, recent_turn_window: *recent_turn_window, prompt: prompt.clone(), @@ -771,10 +772,14 @@ impl RouteConfig { response_schema, policy, )?; + if *classify_trigger != ClassifyTrigger::EveryRequest { + return Err(ServerError::new(format!( + "llm_classifier route {route_name} mode escalation cannot use classify_trigger" + ))); + } if mode.is_some() && (base_threshold.is_some() || threshold_step.is_some() - || *session_affinity || *message_hash_fallback || recent_turn_window.is_some()) { @@ -828,7 +833,7 @@ impl RouteConfig { response_schema, )?, policy: required_classifier_field(route_name, "policy", policy)?, - session_affinity: *session_affinity, + classify_trigger: *classify_trigger, message_hash_fallback: *message_hash_fallback, recent_turn_window: *recent_turn_window, max_output_tokens: *max_output_tokens, @@ -975,7 +980,7 @@ fn build_algorithm( let classifier_config = TaskClassifierConfig { base_threshold: config.base_threshold, threshold_step: config.threshold_step, - session_affinity: config.session_affinity, + classify_trigger: config.classify_trigger, message_hash_fallback: config.message_hash_fallback, recent_turn_window: config.recent_turn_window, contract: classifier_contract(config.prompt.as_deref()) @@ -1024,7 +1029,7 @@ fn build_algorithm( response_schema, config.policy.into_libsy(), ); - classifier_config.session_affinity = config.session_affinity; + classifier_config.classify_trigger = config.classify_trigger; 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; @@ -1496,7 +1501,14 @@ classifier_magic = true "base_threshold = 0.5", "base_threshold = 0.5\nmessage_hash_fallback = true", ), - "message_hash_fallback requires session_affinity", + "message_hash_fallback requires classify_trigger = new_session", + ), + ( + VALID_CONFIG.replace( + "base_threshold = 0.5", + "escalation = { confirmations = 2 }\nclassify_trigger = \"user_turn\"", + ), + "mode escalation cannot use classify_trigger", ), ( VALID_CONFIG.replace("schema_version = 1", "schema_version = 2"), @@ -1612,10 +1624,10 @@ target = "azure" } #[test] - fn accepts_session_affinity_with_message_hash_fallback() -> ServerResult<()> { + fn accepts_new_session_trigger_with_message_hash_fallback() -> ServerResult<()> { let configured = VALID_CONFIG.replace( "base_threshold = 0.5", - "base_threshold = 0.25\nthreshold_step = 0.1\nsession_affinity = true\nmessage_hash_fallback = true", + "base_threshold = 0.25\nthreshold_step = 0.1\nclassify_trigger = \"new_session\"\nmessage_hash_fallback = true", ); server_state_from_toml(&configured)?; Ok(()) diff --git a/dev-server/config.toml b/dev-server/config.toml index 321789081..8455601d5 100644 --- a/dev-server/config.toml +++ b/dev-server/config.toml @@ -74,6 +74,6 @@ classifier_target = "capable" strong_target = "capable" weak_target = "efficient" base_threshold = 0.5 -session_affinity = true +classify_trigger = "new_session" message_hash_fallback = true diff --git a/docs/reference/toml_schema.md b/docs/reference/toml_schema.md index 68d01bc54..27079e37b 100644 --- a/docs/reference/toml_schema.md +++ b/docs/reference/toml_schema.md @@ -154,8 +154,8 @@ Capability mode classifies before serving. See | `weak_target` | Yes | — | Efficient tier. | | `base_threshold` | Yes | — | Lowest solve probability that routes to the weak target. In `[0, 1]`. | | `threshold_step` | No | `0.0` | Finite, non-negative amount added once for uncertain or unmatched verdicts and twice for unsupported verdicts. `base_threshold + 2 * threshold_step` must be at most `1`. | -| `session_affinity` | No | `false` | Reuses a session's first decision on later turns. | -| `message_hash_fallback` | No | `false` | Keys affinity on the first user message. Requires `session_affinity`. | +| `classify_trigger` | No | `every_request` | When the judge runs. `every_request` judges every request, tool continuations included. `user_turn` judges each new user message and holds that target across the tool calls between. `new_session` judges once and reuses that target for the session. | +| `message_hash_fallback` | No | `false` | Keys affinity on the first user message. Requires `classify_trigger = "new_session"`. | | `recent_turn_window` | No | unset | When unset, the judge sees the opening task and latest user follow-up, when present. When set, it also sees trailing turns. | | `prompt` | No | packaged prompt | Replaces the capability prompt. The packaged schema is sent separately as structured-output configuration. | @@ -183,8 +183,8 @@ policy selector, and routes to any configured target label. | `prompt` | Yes | — | Judge system prompt. The configured inner schema is sent separately as structured-output configuration. | | `response_schema` | Yes | — | Inner JSON Schema encoded as a TOML string. Switchyard adds the provider wrapper. | | `policy` | Yes | — | Policy table. `target_selector` accepts a JSON Pointer such as `/decision/target`. | -| `session_affinity` | No | `false` | Reuses a session's first decision on later turns. | -| `message_hash_fallback` | No | `false` | Keys affinity on the first user message. Requires `session_affinity`. | +| `classify_trigger` | No | `every_request` | When the judge runs. `every_request` judges every request, tool continuations included. `user_turn` judges each new user message and holds that target across the tool calls between. `new_session` judges once and reuses that target for the session. | +| `message_hash_fallback` | No | `false` | Keys affinity on the first user message. Requires `classify_trigger = "new_session"`. | | `recent_turn_window` | No | unset | When unset, the judge sees the opening task and latest user follow-up, when present. When set, it also sees trailing turns. | Classifier prompts must not contain `{{RESPONSE_SCHEMA}}`. Switchyard supplies @@ -206,6 +206,7 @@ optional `handoff_notes` and `classifier` tables and for tuning. | `recent_turn_window` | No | `3` | Trailing tool results the signals are computed over. | | `capable_system_prompt` | No | unset | System prompt handed to the capable tier. | | `efficient_system_prompt` | No | unset | System prompt handed to the efficient tier. | +| `classifier.classify_trigger` | No | `every_request` | When the judge runs. See the `llm_classifier` route. `new_session` has no effect here. | | `classifier.response_format_type` | No | `json_schema` | Structured-output mode for the optional classifier judge. Use `json_object` when the classifier provider does not support JSON Schema; Switchyard adds the schema to the prompt and validates the verdict locally. | ## Validation Errors diff --git a/docs/routing_algorithms/llm_classifier_routing.md b/docs/routing_algorithms/llm_classifier_routing.md index a1cc43e45..4706175e3 100644 --- a/docs/routing_algorithms/llm_classifier_routing.md +++ b/docs/routing_algorithms/llm_classifier_routing.md @@ -38,7 +38,7 @@ strong_target = "strong" weak_target = "weak" base_threshold = 0.5 threshold_step = 0.1 -session_affinity = true +classify_trigger = "new_session" message_hash_fallback = true ``` @@ -106,8 +106,8 @@ for the server merge behavior. | `base_threshold` | required | Lowest `p_solve` that routes a supported task to `weak_target`. Must be between `0` and `1`. | | `threshold_step` | `0.0` | Amount added for each boundary step. Must be finite and non-negative, and `base_threshold + 2 * threshold_step` must not exceed `1`. | | `recent_turn_window` | unset | When unset, the judge sees the opening user task and the latest user message when they differ. When set to `N`, it sees the opening user task and the last `N` conversation messages after that task. `0` keeps only the opening task. Client system and developer instructions are not shown to the judge. | -| `session_affinity` | `false` | Retains the first selected target for a session and reuses it on later requests. | -| `message_hash_fallback` | `false` | When session metadata is absent, keys affinity from the first user-message text. Requires `session_affinity = true`. | +| `classify_trigger` | `every_request` | When the judge runs. `every_request` judges every request, tool continuations included. `user_turn` judges each new user message and holds that target across the tool calls between. `new_session` judges once and reuses that target for the session. | +| `message_hash_fallback` | `false` | When session metadata is absent, keys affinity from the first user-message text. Requires `classify_trigger = "new_session"`. | | `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`. | @@ -201,17 +201,34 @@ If a client sends only a follow-up fragment without the opening task, enable affinity or include the task history. Threshold tuning changes routing policy; it cannot recover missing task context. -## Session affinity +## When the judge runs -With `session_affinity = true`, the first selected target is retained for the -request's session identity. This includes `strong_target` when it was selected -as the fallback for an unavailable or unusable judge verdict. There is no -warmup period. Later requests with the same identity reuse the target before -classification, so the judge call is skipped. +`classify_trigger` sets how often the target is re-decided. -Affinity is process-local. Clients can send `x-switchyard-session-id`, or enable -`message_hash_fallback` to key requests without session metadata from the first -user-message text. +`every_request`, the default, judges every request. In an agentic session that +includes every tool continuation, so twenty tool steps means twenty-one +classifications of one task, and the target can change between any two of them. + +`user_turn` judges each new user message and holds that target through the tool +calls that follow: + +```toml +[routes.smart] +classify_trigger = "user_turn" +``` + +Tool results are the agent continuing work the user already asked for, so they +do not re-open the decision. A failed or unusable verdict keeps the current +target. When no target has been selected yet, the next request is judged again. + +`new_session` judges once and reuses that target for the rest of the session, +including `strong_target` when it was selected as the fallback for an unusable +verdict. There is no warmup period, and later requests skip the judge entirely. + +The selection is held in per-session state, so requests without a session +identity are judged every time. Clients can send `x-switchyard-session-id`, or +enable `message_hash_fallback` to key on the first user-message text under +`new_session`. ## Run the route diff --git a/switchyard/cli/defaults/openrouter.toml b/switchyard/cli/defaults/openrouter.toml index 927777d4a..b598511d1 100644 --- a/switchyard/cli/defaults/openrouter.toml +++ b/switchyard/cli/defaults/openrouter.toml @@ -28,5 +28,5 @@ strong_target = "strong" weak_target = "weak" base_threshold = 0.5 threshold_step = 0.0 -session_affinity = true +classify_trigger = "new_session" message_hash_fallback = true