feat(relay): add escalation and stage router support

CHANGELOG.md

@@ -9,10 +9,10 @@ adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 ### Added
 
 - **NeMo Relay native plugin** — a dynamically loaded integration that runs
-  libsy's weighted-random and LLM-classifier algorithms in process while
-  Switchyard owns provider HTTP dispatch, credentials, translation, retries,
-  and fallback. Managed calls require NeMo Relay 0.7 or newer and do not depend
-  on `switchyard-server`.
+  libsy's weighted-random, LLM-classifier, escalation, and stage-router
+  algorithms in process while Switchyard owns provider HTTP dispatch,
+  credentials, translation, retries, and fallback. Managed calls require NeMo
+  Relay 0.7 or newer and do not depend on `switchyard-server`.
 
 ### Removed
 
@@ -39,6 +39,14 @@ adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
 ### Fixed
 
+- **NeMo Relay stage and escalation integration** — preserved same-protocol
+  request bodies are now re-encoded after tier prompts or handoff notes mutate
+  the normalized request; Relay's synthetic `gateway-gateway` identity no
+  longer shares escalation latch state across unrelated raw gateway requests;
+  stage decision marks now retain picker-default tiers, decision sources, and
+  hard-override confidence. Target bindings also accept non-secret
+  `extra_body` defaults for provider-specific judge controls.
+
 - **Response `model` now names the model that actually served the request**, on
   every serving path and wire format. Streamed Anthropic and Responses replies,
   and every libsy-served reply, previously echoed the model id the client

README.md

@@ -21,7 +21,7 @@ algorithm you write yourself.
 - **Protocol Translation**: convert between OpenAI Chat, Anthropic Messages, and OpenAI Responses formats
 - **Multi-Backend Routing**: random routing, LLM-as-classifier routing, signal-driven stage-router, or your own algorithm
 - **Operational Metrics**: Prometheus metrics cover requests, errors, latency, tokens, and routing overhead
-- **NeMo Relay Plugin**: run random or LLM-classifier routing in Relay while Switchyard owns provider HTTP dispatch
+- **NeMo Relay Plugin**: run random, classifier, escalation, or stage routing in Relay while Switchyard owns provider HTTP dispatch
 
 ## Quick Start
 

crates/libsy/src/algorithms/fall_through.rs

@@ -55,6 +55,14 @@ pub struct FallThroughDecision {
     /// Human-readable explanation of the selection.
     pub reasoning: String,
     tier: Option<&'static str>,
+    decision_source: Option<String>,
+}
+
+impl FallThroughDecision {
+    /// Cascade member that resolved the turn, when the assembled router records one.
+    pub fn decision_source(&self) -> Option<&str> {
+        self.decision_source.as_deref()
+    }
 }
 
 impl Decision for FallThroughDecision {
@@ -119,6 +127,7 @@ impl<S: Send> Classifier<S> for DefaultTarget {
 pub struct FallThrough<S = ()> {
     name: String,
     decision_reason: fn(&str, &Score) -> String,
+    decision_source: fn(&S) -> Option<String>,
     processors: Vec<Arc<dyn Processor<S>>>,
     classifiers: Vec<Arc<dyn Classifier<S>>>,
     targets: LlmTargetSet,
@@ -133,6 +142,7 @@ impl FallThrough<()> {
         Self {
             name: "fall_through".to_string(),
             decision_reason: default_decision_reason,
+            decision_source: no_decision_source,
             processors: Vec::new(),
             classifiers: Vec::new(),
             targets,
@@ -152,6 +162,7 @@ where
         Self {
             name: "fall_through".to_string(),
             decision_reason: default_decision_reason,
+            decision_source: no_decision_source,
             processors: Vec::new(),
             classifiers: Vec::new(),
             targets,
@@ -173,6 +184,12 @@ where
         self
     }
 
+    /// Extracts an assembled router's decision source from its composition state.
+    pub(crate) fn with_decision_source(mut self, source: fn(&S) -> Option<String>) -> Self {
+        self.decision_source = source;
+        self
+    }
+
     /// Appends a processor to the head-of-request chain.
     pub fn with_processor(mut self, processor: Arc<dyn Processor<S>>) -> Self {
         self.processors.push(processor);
@@ -286,6 +303,7 @@ where
                 .classifiers
                 .iter()
                 .find_map(|c| c.routing_tier(&to.semantic_name)),
+            decision_source: None,
         })
     }
 
@@ -352,6 +370,7 @@ where
             selected_model: target.semantic_name.clone(),
             reasoning,
             tier: deciding.and_then(|c| c.routing_tier(&target.semantic_name)),
+            decision_source: (self.decision_source)(state),
         });
         driver.info(ctx.clone(), decision.clone()).await?;
 
@@ -411,6 +430,10 @@ fn default_decision_reason(_name: &str, winner: &Score) -> String {
     )
 }
 
+fn no_decision_source<S>(_state: &S) -> Option<String> {
+    None
+}
+
 #[async_trait]
 impl<S> Algorithm for FallThrough<S>
 where

crates/libsy/src/algorithms/stage.rs

@@ -41,12 +41,15 @@ const STAGE_ROUTER: &str = "stage_router";
 struct SourceStamp {
     inner: Arc<dyn Classifier<State>>,
     source: DecisionSource,
+    targets: StageTargets,
 }
 
 #[async_trait]
 impl Classifier<State> for SourceStamp {
     fn routing_tier(&self, selected_model: &str) -> Option<&'static str> {
-        self.inner.routing_tier(selected_model)
+        self.inner
+            .routing_tier(selected_model)
+            .or_else(|| self.targets.label_for(selected_model))
     }
 
     async fn score(
@@ -176,7 +179,8 @@ fn build_route(
     // classifier is a constant rather than a per-turn lookup.
     let fall_open = targets.name(config.mode.default_tier()).to_string();
 
-    let mut classifier = StageClassifier::new(targets, config.mode, config.confidence_threshold);
+    let mut classifier =
+        StageClassifier::new(targets.clone(), config.mode, config.confidence_threshold);
     if let Some(notes) = config.handoff_notes {
         classifier = classifier.with_handoff_notes(notes);
     }
@@ -187,6 +191,7 @@ fn build_route(
     let target_set = LlmTargetSet::new(vec![capable.clone(), efficient.clone()]);
     let mut router = FallThrough::<State>::new_with_state(target_set)
         .with_name(STAGE_ROUTER)
+        .with_decision_source(decision_source)
         .with_processor(Arc::new(signals))
         .with_classifier(Arc::new(classifier));
     if let Some(fallback) = config.llm_fallback {
@@ -200,13 +205,15 @@ fn build_route(
                 config: fallback.config,
             })?),
             source: DecisionSource::LlmClassifier,
+            targets: targets.clone(),
         }));
     }
     // Nothing behind this, so the turn lands on the picker's default tier —
     // including when the judge could not tell.
     router = router.with_classifier(Arc::new(SourceStamp {
         inner: Arc::new(DefaultTarget::new(fall_open)),
         source: DecisionSource::FallOpen,
+        targets,
     }));
     // Runs on the post-decision hook, so it applies to the target the cascade
     // settled on, whichever classifier picked it. With no prompts configured it
@@ -215,6 +222,13 @@ fn build_route(
     Ok(router)
 }
 
+fn decision_source(state: &State) -> Option<String> {
+    match state.extra.get(super::util::stage::DECISION_SOURCE_KEY) {
+        Some(crate::core::state::StateValue::String(source)) => Some(source.clone()),
+        _ => None,
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use std::sync::Arc;
@@ -282,6 +296,7 @@ mod tests {
         let stamp = SourceStamp {
             inner,
             source: DecisionSource::LlmClassifier,
+            targets: StageTargets::new("strong", "weak"),
         };
         let mut state = State::default();
         stamp

crates/libsy/src/algorithms/util/prompts.rs

@@ -41,6 +41,7 @@ pub fn append_note(request: &mut Request, note: &str) {
             .messages
             .push(Message::text(Role::User, note)),
     }
+    request.llm_request.preservation.requests.clear();
 }
 
 /// System prompts keyed by routing target. A target left unset is routed
@@ -104,14 +105,18 @@ impl<S: Send> Processor<S> for SystemPromptProcessor {
                 }],
             },
         );
+        request.llm_request.preservation.requests.clear();
         Ok(())
     }
 }
 
 #[cfg(test)]
 mod tests {
     use super::*;
-    use switchyard_protocol::{LlmRequest, ToolResult, text_request};
+    use serde_json::json;
+    use switchyard_protocol::{
+        LlmRequest, PreservationMetadata, ToolResult, WireFormat, text_request,
+    };
 
     const NOTE: &str = "recovering from an error";
     const STRONG_PROMPT: &str = "diagnose before you edit";
@@ -128,6 +133,16 @@ mod tests {
         }
     }
 
+    fn preserve_chat_request(request: &mut Request) {
+        request.llm_request.preservation = PreservationMetadata {
+            requests: BTreeMap::from([(
+                WireFormat::OpenAiChat.into(),
+                json!({"model": "route", "messages": [{"role": "user", "content": "hi"}]}),
+            )]),
+            ..PreservationMetadata::default()
+        };
+    }
+
     #[test]
     fn a_note_joins_a_trailing_user_turn_after_its_tool_result() {
         // The shape a coding-agent turn actually arrives in: the tool result
@@ -190,6 +205,16 @@ mod tests {
         assert_eq!(trail, vec![format!("fix the build|{NOTE}")]);
     }
 
+    #[test]
+    fn a_note_invalidates_the_preserved_provider_request() {
+        let mut request = request_with(vec![Message::text(Role::User, "fix the build")]);
+        preserve_chat_request(&mut request);
+
+        append_note(&mut request, NOTE);
+
+        assert!(request.llm_request.preservation.requests.is_empty());
+    }
+
     /// A decision routed to `target`.
     struct RoutedTo(&'static str);
     impl switchyard_protocol::Decision for RoutedTo {
@@ -252,6 +277,26 @@ mod tests {
         Ok(())
     }
 
+    #[tokio::test]
+    async fn a_tier_prompt_invalidates_the_preserved_provider_request() -> Result<()> {
+        let processor = SystemPromptProcessor::new(prompts());
+        let mut request = Request::default();
+        preserve_chat_request(&mut request);
+
+        processor
+            .process(
+                &mut (),
+                Event::Decision {
+                    request: &mut request,
+                    decision: &RoutedTo("strong"),
+                },
+            )
+            .await?;
+
+        assert!(request.llm_request.preservation.requests.is_empty());
+        Ok(())
+    }
+
     #[tokio::test]
     async fn an_unconfigured_target_is_left_untouched() -> Result<()> {
         // One target's prompt must not leak onto another, whatever ran before.

crates/libsy/src/algorithms/util/stage.rs

@@ -499,6 +499,7 @@ impl Classifier<State> for StageClassifier {
                 tier,
                 source,
                 score,
+                confidence,
                 ..
             } => {
                 let target = self.targets.name(tier);
@@ -507,7 +508,7 @@ impl Classifier<State> for StageClassifier {
                 // is the only branch whose tier the signals actually chose — an
                 // ambiguous turn is decided further down the cascade.
                 self.apply_handoff_note(request, tier, source);
-                let conf = score.abs();
+                let conf = confidence.unwrap_or_else(|| score.abs());
                 Ok((
                     Classification::Scores(vec![Score {
                         target: target.to_string(),
@@ -643,6 +644,7 @@ mod tests {
             Classification::Scores(scores) => {
                 assert_eq!(scores.len(), 1);
                 assert_eq!(scores[0].target, "strong");
+                assert_eq!(scores[0].confidence, 1.0);
             }
             _ => panic!("expected a definite classification"),
         }

crates/libsy/src/lib.rs

@@ -17,6 +17,7 @@ mod error;
 pub use error::{DriverError, LibsyError, Result};
 
 mod algorithms;
+pub use algorithms::fall_through::FallThroughDecision;
 pub use algorithms::llm_class::{
     CustomClassifierConfig, CustomClassifierPolicy, LlmClassifierConfig, LlmTaskClassifier,
     TaskClassifierConfig,