fix(composite): reset stage routing on user turns

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<Self> {
         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 {

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<dyn Classifier>,
     source: DecisionSource,
     judge: Option<switchyard_protocol::ModelId>,
+    skip_user_turn: bool,
 }
 
 #[async_trait]
@@ -86,6 +87,9 @@ impl Classifier for SourceStamp {
         request: &mut Request,
         driver: &Driver,
     ) -> Result<(Classification, Option<Response>)> {
+        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<Self> {
         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<FallThrough> {
+pub(crate) fn build_stage_route(
+    config: StageRouterConfig,
+    reset_on_user_turn: bool,
+) -> Result<FallThrough> {
     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<FallThrough
     let mut classifier = StageClassifier::new(config.mode, config.confidence_threshold)
         .with_capable_first_threshold(config.capable_first_threshold)
         .with_capable_hold_turns(config.capable_hold_turns);
+    classifier.reset_on_user_turn = reset_on_user_turn;
     if let Some(turns) = config.efficient_hold_turns {
         classifier = classifier.with_efficient_hold_turns(turns);
     }
@@ -301,13 +309,15 @@ pub(crate) fn build_stage_route(config: StageRouterConfig) -> Result<FallThrough
             })?),
             source: DecisionSource::LlmClassifier,
             judge: fallback.target,
+            skip_user_turn: reset_on_user_turn,
         }));
     }
     // Nothing behind this, so no turn is left unrouted.
     router = router.with_classifier(Arc::new(SourceStamp {
         inner: Arc::new(fall_open),
         source: DecisionSource::FallOpen,
         judge: None,
+        skip_user_turn: false,
     }));
     if config.capable_system_prompt.is_some() || config.efficient_system_prompt.is_some() {
         router = router.with_processor(Arc::new(TierPromptProcessor {
@@ -375,6 +385,7 @@ mod tests {
             inner,
             source: DecisionSource::LlmClassifier,
             judge: None,
+            skip_user_turn: false,
         };
         let (driver, mut rx) = Driver::new("test", Arc::new(crate::RuntimeModels::default()));
         stamp
@@ -760,8 +771,9 @@ mod tests {
         let recorder = Arc::new(Recorder::default());
         // The picker would fall open to "strong"; the override says "weak".
         let config = StageRouterConfig::new(PickerMode::CapableFirst, 0.5);
-        let route: Arc<dyn Algorithm> =
-            Arc::new(build_stage_route(config)?.with_processor(Arc::new(TierDecider::default())));
+        let route: Arc<dyn Algorithm> = 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] {

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<i64>,
+    pub(crate) reset_on_user_turn: bool,
     signals: ToolSignalProcessor,
     handoff_notes: Option<HandoffNoteConfig>,
 }
@@ -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 })

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,16 +1045,24 @@ fn extract_tool_signals_with_window(request: &Request, recent_window: usize) ->
         request,
         recent_window,
         &ToolSemantics::default(),
+        false,
     )
 }
 
 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 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<ObservedToolCall> = Vec::new();
@@ -1051,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 `<namespace>__<tool>`.
                     let bare_name = namespaces
@@ -1242,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,
     );
@@ -1786,6 +1809,17 @@ mod tests {
         }
     }
 
+    #[test]
+    fn user_turn_signals_exclude_old_errors_and_keep_user_role_tool_results() {
+        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]
     fn clean_text_has_zero_severity() {
         let (sev, patterns) = classify_text("everything went fine");
@@ -2545,9 +2579,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]

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"

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