fix(libsy): harden the classifier judge path

crates/libsy/src/algorithms/llm_class.rs

@@ -5,6 +5,7 @@
 
 use std::collections::{BTreeMap, BTreeSet, HashSet};
 use std::sync::Arc;
+use std::time::Duration;
 
 use async_trait::async_trait;
 use serde::{Deserialize, Deserializer};
@@ -32,6 +33,17 @@ const SCHEMA_TEMPLATE: &str = include_str!("../prompts/capability-classifier/sch
 /// Telemetry label for this algorithm's spans, metrics, and logs.
 const ALGORITHM_NAME: &str = "llm_task_classifier";
 
+/// Restates the task after the conversation the judge is asked to route.
+///
+/// The rubric leads the request, which works while the payload is a single short
+/// message. Once `recent_turn_window` includes real conversation, those instructions
+/// sit far from the generation point and the judge sometimes *answers* the
+/// conversation instead of classifying it — the reply then fails to parse, the
+/// verdict is unavailable, and the turn silently falls back. Repeating the task at
+/// the end is what reliably prevents that.
+const END_POSITION_REINFORCEMENT: &str =
+    "Route the conversation above. Output ONLY the routing JSON object, nothing else.";
+
 #[derive(Deserialize)]
 #[serde(deny_unknown_fields)]
 struct TaskClassifierVerdict {
@@ -155,10 +167,20 @@ impl ClassifierInput for TaskInput {
     fn build_messages(&self, _state: &State, request: &Request) -> Vec<Message> {
         // The default preserves the whole-task anchor and latest user update. A
         // configured window widens that to the surrounding conversation.
-        match self.recent_turn_window {
+        let mut messages = match self.recent_turn_window {
             Some(window) => trim_messages(&request.llm_request.messages, window),
             None => task_messages(&request.llm_request.messages),
+        };
+        // Only the windowed path carries assistant turns and tool traffic for the judge
+        // to be distracted by. The default path is user task messages only — the anchor
+        // and the latest follow-up — so there is nothing there to outrank.
+        if self.recent_turn_window.is_some() {
+            messages.push(Message::text(
+                Role::User,
+                END_POSITION_REINFORCEMENT.to_string(),
+            ));
         }
+        messages
     }
 }
 
@@ -244,6 +266,13 @@ pub struct TaskClassifierConfig {
     pub contract: ClassifierContractConfig,
     /// Maximum completion tokens available to the classifier verdict.
     pub max_output_tokens: u64,
+    /// Bounds one judge consultation, in milliseconds.
+    ///
+    /// The judge call sits in front of the routed call, so a stalled judge stalls the
+    /// turn. On expiry the judge counts as unavailable and routing falls back exactly
+    /// as it does for any other judge failure. `None` (the default) leaves it
+    /// unbounded, preserving existing behaviour.
+    pub judge_deadline_ms: Option<u64>,
 }
 
 /// Flat serialized shape that maps prompt settings into the runtime contract.
@@ -263,6 +292,8 @@ struct TaskClassifierConfigWire {
     prompt: Option<String>,
     #[serde(default = "default_judge_max_output_tokens")]
     max_output_tokens: u64,
+    #[serde(default)]
+    judge_deadline_ms: Option<u64>,
 }
 
 impl<'de> Deserialize<'de> for TaskClassifierConfig {
@@ -283,6 +314,7 @@ impl<'de> Deserialize<'de> for TaskClassifierConfig {
             recent_turn_window: wire.recent_turn_window,
             contract,
             max_output_tokens: wire.max_output_tokens,
+            judge_deadline_ms: wire.judge_deadline_ms,
         })
     }
 }
@@ -301,6 +333,7 @@ impl Default for TaskClassifierConfig {
             recent_turn_window: None,
             contract: ClassifierContractConfig::default(),
             max_output_tokens: DEFAULT_JUDGE_MAX_OUTPUT_TOKENS,
+            judge_deadline_ms: None,
         }
     }
 }
@@ -694,7 +727,8 @@ impl LlmTaskClassifier {
                     },
                     contract,
                     SerdeDecoder::new(),
-                    JudgeRuntimeConfig::new(config.max_output_tokens)?,
+                    JudgeRuntimeConfig::new(config.max_output_tokens)?
+                        .with_deadline(config.judge_deadline_ms.map(Duration::from_millis))?,
                 ),
                 judge_target.clone(),
                 TaskClassifierPolicy::new(
@@ -1602,6 +1636,50 @@ mod tests {
         );
     }
 
+    #[test]
+    fn a_window_reinforces_the_task_after_the_conversation() -> Result<()> {
+        let contents = judged_contents(2)?;
+
+        // Last, not merely present: the reinforcement only works end-positioned,
+        // after the conversation content it is meant to outrank.
+        assert_eq!(
+            contents.last().map(String::as_str),
+            Some(END_POSITION_REINFORCEMENT)
+        );
+        Ok(())
+    }
+
+    #[test]
+    fn the_single_message_path_is_left_unreinforced() -> Result<()> {
+        // No window means no conversation to be distracted by, so the default
+        // request shape stays exactly as it was.
+        let judge = capability_judge(None)?;
+        let request = Request {
+            llm_request: LlmRequest {
+                messages: vec![Message::text(Role::User, "the task")],
+                ..LlmRequest::default()
+            },
+            raw_request: None,
+            metadata: None,
+        };
+
+        let built = judge.build_request(&State::default(), &request);
+
+        // The task message alone; the rubric reaches the judge as an instruction block
+        // rather than a message, so nothing was dropped by it not being counted here.
+        assert_eq!(built.llm_request.messages.len(), 1);
+        assert!(!built.llm_request.instructions.is_empty());
+        assert!(
+            !built
+                .llm_request
+                .messages
+                .iter()
+                .filter_map(|message| message.text_content("\n"))
+                .any(|text| text.contains(END_POSITION_REINFORCEMENT))
+        );
+        Ok(())
+    }
+
     #[test]
     fn capability_judge_builds_a_structured_request() -> Result<()> {
         let judge = capability_judge(None)?;

crates/libsy/src/algorithms/util.rs

@@ -6,6 +6,7 @@ pub(crate) mod classifier_contract;
 pub mod escalation;
 pub(crate) mod llm_judge;
 pub(crate) mod prompts;
+pub(crate) mod robustness;
 pub(crate) mod stage;
 pub mod subagent;
 pub(crate) mod target_selector;

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;
@@ -16,6 +17,8 @@ use switchyard_protocol::{
     AggLlmResponse, InstructionBlock, LlmRequest, Message, OutputParams, Role, completion_text,
 };
 
+use super::robustness::{safe_client_error, safe_error_summary};
+
 use super::classifier_contract::ClassifierContract;
 use crate::core::algorithm::{Driver, LlmTarget};
 use crate::core::classifier::{Classification, Classifier};
@@ -93,6 +96,8 @@ impl VerdictDecoder for JsonSchemaDecoder {
 /// Runtime limits shared by structured classifier judges.
 pub(crate) struct JudgeRuntimeConfig {
     max_output_tokens: u64,
+    /// Bounds one consultation. `None` leaves it unbounded.
+    deadline: Option<Duration>,
 }
 
 impl JudgeRuntimeConfig {
@@ -102,7 +107,26 @@ impl JudgeRuntimeConfig {
                 message: "max_output_tokens must be at least 1".to_string(),
             });
         }
-        Ok(Self { max_output_tokens })
+        Ok(Self {
+            max_output_tokens,
+            deadline: None,
+        })
+    }
+
+    /// Bounds how long one consultation may take before the judge counts as
+    /// unavailable.
+    ///
+    /// The judge call sits in front of the routed call, so a stalled judge stalls the
+    /// turn behind it with no bound of its own. Expiry folds into the same fail-open
+    /// path as any other judge failure, so the turn is still served.
+    pub(crate) fn with_deadline(mut self, deadline: Option<Duration>) -> Result<Self> {
+        if deadline.is_some_and(|d| d.is_zero()) {
+            return Err(LibsyError::AlgorithmError {
+                message: "judge_deadline_ms must be greater than 0".to_string(),
+            });
+        }
+        self.deadline = deadline;
+        Ok(self)
     }
 }
 
@@ -140,6 +164,10 @@ where
     I: ClassifierInput,
     D: VerdictDecoder,
 {
+    fn deadline(&self) -> Option<Duration> {
+        self.runtime.deadline
+    }
+
     type Verdict = D::Verdict;
 
     fn build_request(&self, state: &State, request: &Request) -> Request {
@@ -178,6 +206,13 @@ pub trait Judge: Send + Sync {
     fn parse(&self, response: &AggLlmResponse) -> Result<Self::Verdict> {
         parse_json_verdict(response)
     }
+
+    /// Bounds one consultation, when the judge is configured with a deadline.
+    ///
+    /// Defaulted so a judge that does not care about liveness need not implement it.
+    fn deadline(&self) -> Option<Duration> {
+        None
+    }
 }
 
 /// Converts a parsed verdict, or an unavailable verdict, into a routing classification.
@@ -225,33 +260,71 @@ where
     ) -> Option<J::Verdict> {
         let judge_model = self.target.semantic_name.as_str();
 
-        let response = driver
-            .call_model(
-                self.judge.build_request(state, request),
-                Decision::new(
-                    self.target.semantic_name.to_string(),
-                    Some("llm judge consultation".to_string()),
-                    false,
-                ),
-            )
-            .await
-            .inspect_err(|error| report_fail_open(judge_model, error, libsy_error_reason(error)))
-            .ok()?;
-        let aggregate = response
-            .llm_response
-            .into_agg()
-            .await
-            .inspect_err(|error| report_fail_open(judge_model, error, client_error_reason(error)))
-            .ok()?;
+        // The whole consultation is bounded, not just the HTTP call: a judge that
+        // returns its headers promptly and then stalls mid-stream would otherwise hold
+        // the turn open just as long.
+        let consult = async {
+            let response = driver
+                .call_model(
+                    self.judge.build_request(state, request),
+                    Decision::new(
+                        self.target.semantic_name.to_string(),
+                        Some("llm judge consultation".to_string()),
+                        false,
+                    ),
+                )
+                .await
+                .inspect_err(|error| {
+                    report_fail_open(
+                        judge_model,
+                        safe_error_summary(error),
+                        libsy_error_reason(error),
+                    )
+                })
+                .ok()?;
+            response
+                .llm_response
+                .into_agg()
+                .await
+                .inspect_err(|error| {
+                    report_fail_open(
+                        judge_model,
+                        safe_client_error(error),
+                        client_error_reason(error),
+                    )
+                })
+                .ok()
+        };
+
+        let aggregate = match self.judge.deadline() {
+            Some(deadline) => match tokio::time::timeout(deadline, consult).await {
+                Ok(aggregate) => aggregate,
+                Err(_) => {
+                    // Libsy-authored text with a configured duration: no upstream content.
+                    report_fail_open(
+                        judge_model,
+                        format!("judge deadline of {} ms exceeded", deadline.as_millis()),
+                        "deadline_exceeded",
+                    );
+                    None
+                }
+            },
+            None => consult.await,
+        }?;
         self.judge
             .parse(&aggregate)
-            .inspect_err(|error| report_fail_open(judge_model, error, "parse_error"))
+            .inspect_err(|error| {
+                report_fail_open(judge_model, safe_error_summary(error), "parse_error")
+            })
             .ok()
     }
 }
 
 /// Logs and counts a judge failure with a bounded label that excludes message content.
-fn report_fail_open(judge_model: &str, error: &dyn std::fmt::Display, reason: &'static str) {
+/// `error` must already be redacted: `LlmClientError::UpstreamHttp`'s `Display` interpolates the
+/// raw upstream body, which can quote the conversation back. Callers pass a
+/// `robustness::safe_*` summary rather than the error itself.
+fn report_fail_open(judge_model: &str, error: String, reason: &'static str) {
     tracing::warn!(
         target: "libsy",
         judge_model,
@@ -335,6 +408,24 @@ fn strip_json_fence(text: &str) -> &str {
 
 #[cfg(test)]
 mod tests {
+    use std::time::Duration;
+
+    #[test]
+    fn a_zero_judge_deadline_is_rejected() {
+        let runtime = JudgeRuntimeConfig::new(4096).expect("valid token budget");
+        assert!(runtime.with_deadline(Some(Duration::ZERO)).is_err());
+
+        // A positive deadline, and no deadline at all, both configure cleanly.
+        let runtime = JudgeRuntimeConfig::new(4096).expect("valid token budget");
+        assert!(
+            runtime
+                .with_deadline(Some(Duration::from_millis(50)))
+                .is_ok()
+        );
+        let runtime = JudgeRuntimeConfig::new(4096).expect("valid token budget");
+        assert!(runtime.with_deadline(None).is_ok());
+    }
+
     use super::*;
 
     use futures::StreamExt;