From e8ed2c64e0fd57a7b0bcf436d02db0de5c3ab113 Mon Sep 17 00:00:00 2001 From: Ryan Lempka Date: Fri, 30 Oct 2026 13:16:28 -0500 Subject: [PATCH 1/3] fix(composite): reset stage routing on user turns Signed-off-by: Ryan Lempka --- crates/libsy/src/algorithms/composite.rs | 21 ++++++++++---- crates/libsy/src/algorithms/stage.rs | 22 ++++++++++---- crates/libsy/src/algorithms/util/stage.rs | 14 +++++++++ .../libsy/src/algorithms/util/tool_signals.rs | 29 ++++++++++++++++++- .../tests/composite_profiles.rs | 2 +- docs/routing_algorithms/composite_routing.md | 10 +++---- 6 files changed, 80 insertions(+), 18 deletions(-) diff --git a/crates/libsy/src/algorithms/composite.rs b/crates/libsy/src/algorithms/composite.rs index 6ac0073063..7d22e8a0fb 100644 --- a/crates/libsy/src/algorithms/composite.rs +++ b/crates/libsy/src/algorithms/composite.rs @@ -146,7 +146,7 @@ impl CompositeRouter { /// of the fall-open tier and so answers most of the turns this one set a tier for. pub fn new(config: CompositeRouterConfig) -> Result { let setter = Self::tier_setter(config.judge, config.stage.mode.default_tier())?; - let route = build_stage_route(config.stage)? + let route = build_stage_route(config.stage, setter.trigger == ClassifyTrigger::UserTurn)? .with_name(COMPOSITE) .with_processor(Arc::new(setter)); Ok(Self { route }) @@ -164,8 +164,11 @@ impl CompositeRouter { } } let setter = Self::tier_setter(judge, Tier::Efficient)?; - let profiles = - StageProfiles([build_stage_route(profile_0)?, build_stage_route(profile_1)?]); + let reset_on_user_turn = setter.trigger == ClassifyTrigger::UserTurn; + let profiles = StageProfiles([ + build_stage_route(profile_0, reset_on_user_turn)?, + build_stage_route(profile_1, reset_on_user_turn)?, + ]); let route = FallThrough::new_with_state() .with_name(COMPOSITE) .with_processor(Arc::new(setter)) @@ -414,7 +417,7 @@ mod tests { } #[tokio::test] - async fn profiles_keep_their_own_holds_and_prompts_when_the_judge_switches() -> Result<()> { + async fn profiles_reset_holds_and_keep_prompts_when_the_judge_switches() -> Result<()> { for (has_session_id, picker, first_target, first_prompt) in [ (false, PickerMode::EfficientFirst, "weak", "EF efficient"), (true, PickerMode::EfficientFirst, "weak", "EF efficient"), @@ -428,6 +431,12 @@ mod tests { capable.capable_hold_turns = 0; capable.capable_system_prompt = Some("CF capable".into()); capable.efficient_system_prompt = Some("CF efficient".into()); + for stage in [&mut efficient, &mut capable] { + stage.llm_fallback = Some(crate::algorithms::stage::LlmFallback { + target: None, + config: Default::default(), + }); + } let router = Arc::new(CompositeRouter::new_with_profiles( CompositeJudgeConfig::Capability(TaskClassifierConfig { judge: CapabilityJudgeConfig::Llm(LlmCapabilityConfig { @@ -453,8 +462,8 @@ mod tests { (1.0, turn_request(true), "strong", "EF capable"), (0.0, user_turn_request(), "strong", "CF capable"), (0.0, write, "weak", "CF efficient"), - (1.0, user_turn_request(), "strong", "EF capable"), - (0.0, user_turn_request(), "weak", "CF efficient"), + (1.0, user_turn_request(), first_target, first_prompt), + (0.0, user_turn_request(), "strong", "CF capable"), ] { *recorder.judge_p_solve.lock() = p_solve; let request = if has_session_id { diff --git a/crates/libsy/src/algorithms/stage.rs b/crates/libsy/src/algorithms/stage.rs index 77795593d6..06936a8ed4 100644 --- a/crates/libsy/src/algorithms/stage.rs +++ b/crates/libsy/src/algorithms/stage.rs @@ -19,7 +19,7 @@ use async_trait::async_trait; use super::fall_through::FallThrough; use super::llm_class::{LlmClassifierConfig, LlmTaskClassifier, TaskClassifierConfig}; -use super::util::affinity::ClassifyTrigger; +use super::util::affinity::{ClassifyTrigger, has_new_user_turn}; use super::util::stage::{ DecisionSource, HandoffNoteConfig, PickerMode, StageClassifier, Tier, fall_open_tier, record_routing_decision, @@ -76,6 +76,7 @@ struct SourceStamp { inner: Arc, source: DecisionSource, judge: Option, + skip_user_turn: bool, } #[async_trait] @@ -86,6 +87,9 @@ impl Classifier for SourceStamp { request: &mut Request, driver: &Driver, ) -> Result<(Classification, Option)> { + if self.skip_user_turn && has_new_user_turn(&request.llm_request.messages) { + return Ok((Classification::Ambiguous(Vec::new()), None)); + } let scoped = self .judge .as_ref() @@ -214,7 +218,7 @@ impl StageRouter { /// turn, so `classify_trigger` and `message_hash_fallback` would do nothing. pub fn new(config: StageRouterConfig) -> Result { Ok(Self { - route: build_stage_route(config)?, + route: build_stage_route(config, false)?, }) } } @@ -236,7 +240,10 @@ impl Algorithm for StageRouter { /// Wires the cascade the wrapper drives. Exposed so a composition above can /// stack a prelude onto it. -pub(crate) fn build_stage_route(config: StageRouterConfig) -> Result { +pub(crate) fn build_stage_route( + config: StageRouterConfig, + reset_on_user_turn: bool, +) -> Result { if let Some(fallback) = &config.llm_fallback && (fallback.config.classify_trigger != ClassifyTrigger::EveryRequest || fallback.config.message_hash_fallback) @@ -279,6 +286,7 @@ pub(crate) fn build_stage_route(config: StageRouterConfig) -> Result Result Result = - Arc::new(build_stage_route(config)?.with_processor(Arc::new(TierDecider::default()))); + let route: Arc = Arc::new( + build_stage_route(config, false)?.with_processor(Arc::new(TierDecider::default())), + ); let models = runtime_models(); for is_turn_failed in [false, false, true, false] { diff --git a/crates/libsy/src/algorithms/util/stage.rs b/crates/libsy/src/algorithms/util/stage.rs index 8f0d182319..d607609763 100644 --- a/crates/libsy/src/algorithms/util/stage.rs +++ b/crates/libsy/src/algorithms/util/stage.rs @@ -21,6 +21,7 @@ use async_trait::async_trait; use opentelemetry::KeyValue; use serde::Deserialize; +use super::affinity::has_new_user_turn; use super::prompts; use super::tool_signals::{ToolSignalProcessor, ToolSignals}; use crate::Result; @@ -530,6 +531,7 @@ pub struct StageClassifier { capable_first_threshold: f64, capable_hold_turns: i64, efficient_hold_turns: Option, + pub(crate) reset_on_user_turn: bool, signals: ToolSignalProcessor, handoff_notes: Option, } @@ -544,6 +546,7 @@ impl StageClassifier { capable_first_threshold: 0.0, capable_hold_turns: 2, efficient_hold_turns: None, + reset_on_user_turn: false, signals: ToolSignalProcessor::default(), handoff_notes: None, } @@ -646,6 +649,13 @@ impl StageClassifier { impl Processor for StageClassifier { async fn process(&self, state: &mut State, event: Event<'_>) -> Result<()> { if let Event::Request { request, driver } = event { + if self.reset_on_user_turn && has_new_user_turn(&request.llm_request.messages) { + for key in [CAPABLE_HOLD_KEY, EFFICIENT_HOLD_KEY, PREVIOUS_TIER_KEY] { + state.extra.remove(&Self::agent_key(request, key)); + } + state.tool_signals = None; + return Ok(()); + } let held = [CAPABLE_HOLD_KEY, EFFICIENT_HOLD_KEY].iter().any(|key| { match state.extra.get(&Self::agent_key(request, key)) { Some(StateValue::Bool(true)) => true, @@ -657,6 +667,10 @@ impl Processor for StageClassifier { state.tool_signals = None; return Ok(()); } + if self.reset_on_user_turn { + state.tool_signals = Some(self.signals.for_user_turn(request)); + return Ok(()); + } return self .signals .process(state, Event::Request { request, driver }) diff --git a/crates/libsy/src/algorithms/util/tool_signals.rs b/crates/libsy/src/algorithms/util/tool_signals.rs index 5dda84ef9f..ba5bcd3648 100644 --- a/crates/libsy/src/algorithms/util/tool_signals.rs +++ b/crates/libsy/src/algorithms/util/tool_signals.rs @@ -23,6 +23,7 @@ use serde_json::Value; use switchyard_protocol::codex_namespaces::{split_qualified_name, tool_namespaces}; use switchyard_protocol::{ContentBlock, Request, Role, WireFormat}; +use super::affinity::is_user_turn; use crate::{LibsyError, Result}; use crate::core::processor::{Event, Processor}; @@ -421,6 +422,7 @@ impl ToolSignals { request, window_size.unwrap_or(DEFAULT_RECENT_WINDOW), semantics, + false, ) } } @@ -473,6 +475,17 @@ impl Default for ToolSignalProcessor { } } +impl ToolSignalProcessor { + pub(crate) fn for_user_turn(&self, request: &Request) -> ToolSignals { + extract_tool_signals_with_window_and_semantics( + request, + self.recent_window, + &self.tool_semantics, + true, + ) + } +} + #[async_trait] impl Processor for ToolSignalProcessor { async fn process(&self, state: &mut State, event: Event<'_>) -> Result<()> { @@ -1032,6 +1045,7 @@ fn extract_tool_signals_with_window(request: &Request, recent_window: usize) -> request, recent_window, &ToolSemantics::default(), + false, ) } @@ -1039,9 +1053,13 @@ fn extract_tool_signals_with_window_and_semantics( request: &Request, recent_window: usize, semantics: &ToolSemantics, + current_user_turn: bool, ) -> ToolSignals { // Read the decoded conversation, including preserved built-in tool outputs. - let messages = &request.llm_request.messages; + let mut messages = request.llm_request.messages.as_slice(); + if current_user_turn { + messages = &messages[messages.iter().rposition(is_user_turn).unwrap_or(0)..]; + } let namespaces = tool_namespaces(&request.llm_request.extensions); let mut tool_texts: Vec<(String, bool)> = Vec::new(); let mut tool_calls: Vec = Vec::new(); @@ -1786,6 +1804,15 @@ mod tests { } } + #[test] + fn user_turn_signals_exclude_old_errors_and_keep_user_role_tool_results() { + let user = Message::text(Role::User, "next"); + let request = with_messages(vec![tr("out of memory"), user, tr("ok")]); + let signals = ToolSignalProcessor::default().for_user_turn(&request); + assert_eq!(signals.tool_result_count, 1); + assert_eq!(signals.severity, 0.0); + } + #[test] fn clean_text_has_zero_severity() { let (sev, patterns) = classify_text("everything went fine"); diff --git a/crates/switchyard-runner/tests/composite_profiles.rs b/crates/switchyard-runner/tests/composite_profiles.rs index 84d19a1671..d87d8c78d7 100644 --- a/crates/switchyard-runner/tests/composite_profiles.rs +++ b/crates/switchyard-runner/tests/composite_profiles.rs @@ -33,7 +33,7 @@ type = "composite" [routes.auto.classifier] target = "judge" base_threshold = 0.5 -classify_trigger = "user_turn" +classify_trigger = "new_session" [routes.auto.stage] capable_target = "strong" efficient_target = "weak" diff --git a/docs/routing_algorithms/composite_routing.md b/docs/routing_algorithms/composite_routing.md index c5949acbbb..2a806097b9 100644 --- a/docs/routing_algorithms/composite_routing.md +++ b/docs/routing_algorithms/composite_routing.md @@ -18,8 +18,8 @@ relationship between them are fixed, and the configuration below names them directly rather than composing arbitrary algorithms. Under capable-first, `stage.efficient_hold_turns` defaults to `-1`: after switching -to efficient, it stays there until session state is cleared. Efficient-first -defaults to `0`. Set this field explicitly to override either default; `0` +to efficient, it stays there until the next user-turn reset or session reset. Efficient-first +defaults to `0`. Set this field explicitly to override either default. `0` disables the hold, and a positive value holds that many subsequent requests. The classifier can use an LLM judge or a decision model. A decision @@ -95,9 +95,9 @@ On model fallback, each candidate receives the active profile's prompt for its t Omitted settings inherit from `stage`, including the corresponding threshold. A supplied `tool_semantics` or `handoff_notes` table replaces the shared table. -Either table enables two profiles with separate hold state. Switching back -resumes that profile's holds. Inactive profiles do not consume hold turns. -With neither table, existing behavior is preserved. +With `user_turn`, each new user message resets holds and tool signals and starts +on the selected profile's initial tier. Its classifier runs only on tool continuations. +Signals use only the current user turn. `new_session` retains session-wide state. ## Custom LLM classifier From 93b0b4bbfd723f94d75cbe09f31e400f3dd290c8 Mon Sep 17 00:00:00 2001 From: Ryan Lempka Date: Fri, 30 Oct 2026 13:38:01 -0500 Subject: [PATCH 2/3] fix(composite): preserve tool continuation boundaries Signed-off-by: Ryan Lempka --- crates/libsy/src/algorithms/util/affinity.rs | 28 ++++++++++++---- .../libsy/src/algorithms/util/tool_signals.rs | 33 ++++++++++++++----- 2 files changed, 46 insertions(+), 15 deletions(-) diff --git a/crates/libsy/src/algorithms/util/affinity.rs b/crates/libsy/src/algorithms/util/affinity.rs index 8b3840fa13..cdf27894af 100644 --- a/crates/libsy/src/algorithms/util/affinity.rs +++ b/crates/libsy/src/algorithms/util/affinity.rs @@ -24,7 +24,7 @@ use std::sync::atomic::{AtomicBool, Ordering}; use async_trait::async_trait; use parking_lot::Mutex; use serde::Deserialize; -use switchyard_protocol::{Category, ContentBlock, Message, ModelId, Request, Role}; +use switchyard_protocol::{Category, ContentBlock, Message, ModelId, Request, Role, WireFormat}; use crate::core::algorithm::{Driver, RoutingIdentity}; use crate::core::classifier::{Classification, Classifier, Score}; @@ -48,14 +48,28 @@ pub enum ClassifyTrigger { NewSession, } -/// Anthropic carries tool results as a `Role::User` message, so role alone cannot tell a -/// human turn from a tool continuation. +/// Anthropic and Responses can carry tool activity as user-role messages. pub(crate) fn is_user_turn(message: &Message) -> bool { message.role == Role::User - && !message - .content - .iter() - .all(|block| matches!(block, ContentBlock::ToolResult(_))) + && !message.content.iter().all(|block| match block { + ContentBlock::ToolResult(_) => true, + ContentBlock::Unknown { provider, raw } + if provider.as_str() == WireFormat::OpenAiResponses.as_str() => + { + matches!( + raw["type"].as_str(), + Some( + "shell_call" + | "shell_call_output" + | "apply_patch_call" + | "apply_patch_call_output" + | "computer_call" + | "computer_call_output" + ) + ) + } + _ => false, + }) } pub(crate) fn has_new_user_turn(messages: &[Message]) -> bool { diff --git a/crates/libsy/src/algorithms/util/tool_signals.rs b/crates/libsy/src/algorithms/util/tool_signals.rs index ba5bcd3648..49e968b4e8 100644 --- a/crates/libsy/src/algorithms/util/tool_signals.rs +++ b/crates/libsy/src/algorithms/util/tool_signals.rs @@ -1056,10 +1056,13 @@ fn extract_tool_signals_with_window_and_semantics( current_user_turn: bool, ) -> ToolSignals { // Read the decoded conversation, including preserved built-in tool outputs. - let mut messages = request.llm_request.messages.as_slice(); - if current_user_turn { - messages = &messages[messages.iter().rposition(is_user_turn).unwrap_or(0)..]; - } + let messages = &request.llm_request.messages; + let boundary = if current_user_turn { + messages.iter().rposition(is_user_turn) + } else { + None + }; + let start = boundary.map_or(0, |index| index + 1); let namespaces = tool_namespaces(&request.llm_request.extensions); let mut tool_texts: Vec<(String, bool)> = Vec::new(); let mut tool_calls: Vec = Vec::new(); @@ -1069,12 +1072,14 @@ fn extract_tool_signals_with_window_and_semantics( let mut tool_result_count = 0usize; let mut assistant_turn_count = 0usize; - for message in messages { - if message.role == Role::Assistant { + for (index, message) in messages.iter().enumerate() { + if index >= start && message.role == Role::Assistant { assistant_turn_count += 1; } for block in &message.content { match block { + // Only context-compaction markers carry across user turns. + _ if index < start && !matches!(block, ContentBlock::Text { .. }) => continue, ContentBlock::ToolCall(call) => { // Responses namespaced tools arrive as `__`. let bare_name = namespaces @@ -1260,7 +1265,7 @@ fn extract_tool_signals_with_window_and_semantics( let mut signal = build_signal( tool_texts, tool_calls, - messages.len() as u32, + (messages.len() - boundary.unwrap_or(0)) as u32, recent_window, semantics, ); @@ -1806,11 +1811,13 @@ mod tests { #[test] fn user_turn_signals_exclude_old_errors_and_keep_user_role_tool_results() { - let user = Message::text(Role::User, "next"); + let mut user = Message::text(Role::User, "next"); + user.content.extend(tr("out of memory").content); let request = with_messages(vec![tr("out of memory"), user, tr("ok")]); let signals = ToolSignalProcessor::default().for_user_turn(&request); assert_eq!(signals.tool_result_count, 1); assert_eq!(signals.severity, 0.0); + assert_eq!(signals.turn_depth, 2); } #[test] @@ -2385,6 +2392,10 @@ mod tests { .collect(), ); let signal = ToolSignals::from_request(&request, Some(3)); + assert!(!request.llm_request.messages.iter().any(is_user_turn)); + let current = ToolSignalProcessor::default().for_user_turn(&request); + assert_eq!(current.tool_result_count, 2); + assert_eq!(current.severity, severity); assert_eq!(signal.severity, severity, "{raw}"); assert_eq!(signal.repeated_failure, is_error, "{raw}"); assert_eq!(signal.tool_result_count, 2); @@ -2572,9 +2583,15 @@ mod tests { Role::User, "This session is being continued from a previous conversation that ran out of context.", ), + Message::text(Role::User, "Continue with the next task"), bash("ls"), ]); assert!(ToolSignals::from_request(&request, None).compacted); + assert!( + ToolSignalProcessor::default() + .for_user_turn(&request) + .compacted + ); } #[test] From 1c8eafbbbe64e5607f1595c0aa1022de8fa6a1bf Mon Sep 17 00:00:00 2001 From: Ryan Lempka Date: Fri, 30 Oct 2026 14:14:57 -0500 Subject: [PATCH 3/3] refactor(composite): separate built-in tool normalization Signed-off-by: Ryan Lempka --- crates/libsy/src/algorithms/util/affinity.rs | 28 +++++-------------- .../libsy/src/algorithms/util/tool_signals.rs | 4 --- 2 files changed, 7 insertions(+), 25 deletions(-) diff --git a/crates/libsy/src/algorithms/util/affinity.rs b/crates/libsy/src/algorithms/util/affinity.rs index cdf27894af..8b3840fa13 100644 --- a/crates/libsy/src/algorithms/util/affinity.rs +++ b/crates/libsy/src/algorithms/util/affinity.rs @@ -24,7 +24,7 @@ use std::sync::atomic::{AtomicBool, Ordering}; use async_trait::async_trait; use parking_lot::Mutex; use serde::Deserialize; -use switchyard_protocol::{Category, ContentBlock, Message, ModelId, Request, Role, WireFormat}; +use switchyard_protocol::{Category, ContentBlock, Message, ModelId, Request, Role}; use crate::core::algorithm::{Driver, RoutingIdentity}; use crate::core::classifier::{Classification, Classifier, Score}; @@ -48,28 +48,14 @@ pub enum ClassifyTrigger { NewSession, } -/// Anthropic and Responses can carry tool activity as user-role messages. +/// Anthropic carries tool results as a `Role::User` message, so role alone cannot tell a +/// human turn from a tool continuation. pub(crate) fn is_user_turn(message: &Message) -> bool { message.role == Role::User - && !message.content.iter().all(|block| match block { - ContentBlock::ToolResult(_) => true, - ContentBlock::Unknown { provider, raw } - if provider.as_str() == WireFormat::OpenAiResponses.as_str() => - { - matches!( - raw["type"].as_str(), - Some( - "shell_call" - | "shell_call_output" - | "apply_patch_call" - | "apply_patch_call_output" - | "computer_call" - | "computer_call_output" - ) - ) - } - _ => false, - }) + && !message + .content + .iter() + .all(|block| matches!(block, ContentBlock::ToolResult(_))) } pub(crate) fn has_new_user_turn(messages: &[Message]) -> bool { diff --git a/crates/libsy/src/algorithms/util/tool_signals.rs b/crates/libsy/src/algorithms/util/tool_signals.rs index 49e968b4e8..a5eb175191 100644 --- a/crates/libsy/src/algorithms/util/tool_signals.rs +++ b/crates/libsy/src/algorithms/util/tool_signals.rs @@ -2392,10 +2392,6 @@ mod tests { .collect(), ); let signal = ToolSignals::from_request(&request, Some(3)); - assert!(!request.llm_request.messages.iter().any(is_user_turn)); - let current = ToolSignalProcessor::default().for_user_turn(&request); - assert_eq!(current.tool_result_count, 2); - assert_eq!(current.severity, severity); assert_eq!(signal.severity, severity, "{raw}"); assert_eq!(signal.repeated_failure, is_error, "{raw}"); assert_eq!(signal.tool_result_count, 2);