From 1dbbc016c415a9e7c48669c4000573f05baabef4 Mon Sep 17 00:00:00 2001 From: Joel Fernandes Date: Tue, 11 Aug 2026 14:30:47 -0400 Subject: [PATCH] feat(libsy): warn once when affinity cannot derive a key Signed-off-by: Joel Fernandes --- .../libsy-llm-client/tests/observability.rs | 50 ++++++++- crates/libsy/src/algorithms/util/affinity.rs | 106 +++++++++++++----- 2 files changed, 126 insertions(+), 30 deletions(-) diff --git a/crates/libsy-llm-client/tests/observability.rs b/crates/libsy-llm-client/tests/observability.rs index 18d8f9f3..8e3c6697 100644 --- a/crates/libsy-llm-client/tests/observability.rs +++ b/crates/libsy-llm-client/tests/observability.rs @@ -33,8 +33,9 @@ use tracing_subscriber::layer::{Context as LayerContext, SubscriberExt}; use tracing_subscriber::registry::LookupSpan; use switchyard_libsy::{ - Algorithm, Driver, LibsyError, LlmClassifierConfig, LlmTarget, LlmTargetSet, LlmTaskClassifier, - PickerMode, StageRouter, StageRouterConfig, Step, TaskClassifierConfig, + AffinityRouter, Algorithm, Classifier, Driver, LibsyError, LlmClassifierConfig, LlmTarget, + LlmTargetSet, LlmTaskClassifier, PickerMode, StageRouter, StageRouterConfig, Step, + TaskClassifierConfig, }; use switchyard_llm_client::{ClientRouter, RunObservation, RunObserver}; use switchyard_protocol::{ @@ -570,6 +571,51 @@ fn otel_attribute<'a>(span: &'a SpanData, key: &str) -> Option<&'a OtelValue> { .map(|attribute| &attribute.value) } +#[tokio::test] +async fn affinity_warns_once_when_request_has_no_usable_identity() -> switchyard_libsy::Result<()> { + let _guard = serialize_test().lock().await; + let (store, _, _, _, _) = telemetry(); + let event_count = store.events().len(); + let router = AffinityRouter::new().with_message_hash_fallback(); + let mut state = (); + let mut request = Request { + llm_request: LlmRequest { + messages: vec![Message { + role: Role::User, + content: vec![ContentBlock::Reasoning { + text: "provider reasoning".to_string(), + signature: None, + }], + }], + ..LlmRequest::default() + }, + raw_request: None, + metadata: None, + }; + + for _ in 0..2 { + router.score(&mut state, &mut request, None).await?; + } + + let events = store.events(); + let warnings = events[event_count..] + .iter() + .filter(|event| { + event.target == "libsy" + && event.level == "WARN" + && event + .fields + .get("message") + .is_some_and(|message| message.contains("affinity is enabled")) + }) + .collect::>(); + assert_eq!(warnings.len(), 1, "affinity warnings: {warnings:?}"); + assert!(warnings[0].fields.get("message").is_some_and(|message| { + message.contains("message-hash fallback with usable first-user text") + })); + Ok(()) +} + #[tokio::test] async fn successful_run_records_metrics_spans_and_decision_log() -> switchyard_libsy::Result<()> { let _guard = serialize_test().lock().await; diff --git a/crates/libsy/src/algorithms/util/affinity.rs b/crates/libsy/src/algorithms/util/affinity.rs index 6bcbb681..bef22d17 100644 --- a/crates/libsy/src/algorithms/util/affinity.rs +++ b/crates/libsy/src/algorithms/util/affinity.rs @@ -19,6 +19,7 @@ use std::collections::{HashMap, HashSet, hash_map::DefaultHasher}; use std::hash::{Hash, Hasher}; +use std::sync::atomic::{AtomicBool, Ordering}; use async_trait::async_trait; use parking_lot::Mutex; @@ -55,6 +56,8 @@ pub struct AffinityRouter { /// Held on the instance so the two roles share one process-local map through a /// single registered [`Arc`](std::sync::Arc); bounded by [`MAX_ASSIGNMENTS`]. assignments: Mutex>, + /// Whether the "no identity to key on" warning has already been emitted. + unkeyed_warning_emitted: AtomicBool, } impl AffinityRouter { @@ -95,26 +98,45 @@ impl AffinityRouter { /// Derives the stable identity this router should retain for `request`. fn affinity_key(&self, request: &Request) -> Option { - if let Some(identity) = RoutingIdentity::from_request(request) { - return match identity { - RoutingIdentity::Session(_) if self.subagents_only => None, - identity => Some(identity), - }; + let metadata = request.metadata.as_ref(); + let is_subagent = metadata.is_some_and(|metadata| metadata.is_subagent); + // This mode handles only subagent requests; root requests intentionally fall through. + if self.subagents_only && !is_subagent { + return None; } - // If headers are not present and we are not a subagent, use the message hash based fallback key to do task based routing - let is_subagent = request - .metadata - .as_ref() - .is_some_and(|metadata| metadata.is_subagent); - (!self.subagents_only && !is_subagent && self.message_hash_fallback) - .then(|| { + let key = match (RoutingIdentity::from_request(request), is_subagent) { + (Some(identity), _) => Some(identity), + // Child requests require their explicit session + agent identity and never fall + // back to task text. + (None, true) => None, + (None, false) if self.message_hash_fallback => { first_user_message_hash(request).map(|hash| { tracing::debug!(affinity_key = %hash, "affinity using message hash fallback"); RoutingIdentity::Session(hash) }) - }) - .flatten() + } + (None, false) => None, + }; + // Affinity that never keys anything is silent otherwise: the route reports itself as + // configured while every turn is classified afresh. Say so once rather than per turn. + if key.is_none() && self.should_warn_unkeyed() { + tracing::warn!( + target: "libsy", + is_subagent, + message_hash_fallback = self.message_hash_fallback, + "affinity is enabled but this request carries no usable identity, so no \ + affinity is applied; root requests need a session id or message-hash \ + fallback with usable first-user text, and child requests need both session \ + and agent ids" + ); + } + key + } + + /// Reports whether this call owns the one-time warning for an unkeyable request. + fn should_warn_unkeyed(&self) -> bool { + !self.unkeyed_warning_emitted.swap(true, Ordering::Relaxed) } } @@ -463,6 +485,21 @@ mod tests { Ok(()) } + #[tokio::test] + async fn subagents_only_root_traffic_does_not_warn() -> Result<(), BoxErr> { + // Abstaining on root traffic is this mode's contract, so it must not warn. + let router = AffinityRouter::for_subagents(); + let mut state = (); + + let mut root = request(session("session-1", "agent-1")); + assert!(scores(&router, &mut state, &mut root).await?.is_empty()); + assert!( + router.should_warn_unkeyed(), + "an intentional abstention should leave the warning unconsumed" + ); + Ok(()) + } + #[test] fn user_message_hash_ignores_non_text_provider_payloads() { let request = |user_message| Request { @@ -519,19 +556,28 @@ mod tests { } #[tokio::test] - async fn message_hash_fallback_abstains_for_subagents() -> Result<(), BoxErr> { - let router = AffinityRouter::new().with_message_hash_fallback(); - let mut state = (); - let mut subagent = task_request( - Some(Metadata { - is_subagent: true, - ..Metadata::default() - }), - "Implement the parser.", - None, - ); - - assert!(scores(&router, &mut state, &mut subagent).await?.is_empty()); + async fn subagent_without_a_session_abstains_and_warns() -> Result<(), BoxErr> { + for session_id in [None, Some(String::new())] { + let router = AffinityRouter::new().with_message_hash_fallback(); + let mut state = (); + let mut subagent = task_request( + Some(Metadata { + session_id, + agent_id: Some("agent-1".to_string()), + is_subagent: true, + ..Metadata::default() + }), + "Implement the parser.", + None, + ); + + retain(&router, &mut state, &mut subagent, "model-a").await?; + assert!(scores(&router, &mut state, &mut subagent).await?.is_empty()); + assert!( + !router.should_warn_unkeyed(), + "an unidentifiable subagent should consume the warning" + ); + } Ok(()) } @@ -670,7 +716,7 @@ mod tests { } #[tokio::test] - async fn subagent_without_an_agent_id_is_not_keyed() -> Result<(), BoxErr> { + async fn subagent_without_an_agent_id_abstains_and_warns() -> Result<(), BoxErr> { let router = AffinityRouter::new(); let mut state = (); @@ -684,6 +730,10 @@ mod tests { let mut req = request(metadata); retain(&router, &mut state, &mut req, "model-a").await?; assert!(scores(&router, &mut state, &mut req).await?.is_empty()); + assert!( + !router.should_warn_unkeyed(), + "an unidentifiable subagent should consume the warning" + ); Ok(()) }