diff --git a/crates/libsy/src/algorithms/util/affinity.rs b/crates/libsy/src/algorithms/util/affinity.rs index 04aa5922..47381c7f 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; @@ -67,6 +68,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 { @@ -107,34 +110,51 @@ impl AffinityRouter { /// Derives the stable identity this router should retain for `request`. fn affinity_key(&self, request: &Request) -> Option { - if let Some(metadata) = request.metadata.as_ref() - && let Some(session) = metadata.session_id.clone() - { - return if metadata.is_subagent { - metadata - .agent_id - .clone() - .map(|agent| AffinityKey::Subagent { session, agent }) - } else if self.subagents_only { - None - } else { - Some(AffinityKey::Session(session)) - }; + let metadata = request.metadata.as_ref(); + let is_subagent = metadata.is_some_and(|metadata| metadata.is_subagent); + // Abstaining on root traffic is this mode's contract, not a misconfiguration. + 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(|| { + // An empty session header carries no identity: keying on it would collapse every + // task onto one assignment. Treat it as absent, as the fall-through router does. + let session = metadata + .and_then(|metadata| metadata.session_id.clone()) + .filter(|session| !session.is_empty()); + let key = match (session, is_subagent) { + (Some(session), true) => metadata + .and_then(|metadata| metadata.agent_id.clone()) + .map(|agent| AffinityKey::Subagent { session, agent }), + (Some(session), false) => Some(AffinityKey::Session(session)), + // 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"); AffinityKey::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!( + 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, 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) } } @@ -482,6 +502,69 @@ mod tests { Ok(()) } + #[tokio::test] + async fn empty_session_id_is_treated_as_absent() -> Result<(), BoxErr> { + let router = AffinityRouter::new().with_message_hash_fallback(); + let mut state = (); + // A harness that sets the session header to an empty string, rather than omitting it. + let empty_session = || { + Some(Metadata { + session_id: Some(String::new()), + ..Metadata::default() + }) + }; + + let mut first = task_request(empty_session(), "Add a unit test for this function.", None); + retain(&router, &mut state, &mut first, "weak").await?; + + // An unrelated task shares the empty session id, so keying on it would wrongly + // hand this task the first task's model. + let mut other_task = task_request( + empty_session(), + "Reimplement this binary from two input/output pairs.", + None, + ); + assert!( + scores(&router, &mut state, &mut other_task) + .await? + .is_empty() + ); + Ok(()) + } + + #[tokio::test] + async fn unkeyable_requests_warn_only_once() -> Result<(), BoxErr> { + // Affinity without the fallback and without session metadata can never key anything. + let router = AffinityRouter::new(); + let mut state = (); + + let mut first = task_request(None, "Add a unit test for this function.", None); + assert!(scores(&router, &mut state, &mut first).await?.is_empty()); + assert!( + !router.should_warn_unkeyed(), + "the first unkeyable request should have consumed the warning" + ); + + let mut second = task_request(None, "Reimplement this binary.", None); + assert!(scores(&router, &mut state, &mut second).await?.is_empty()); + 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 { @@ -538,19 +621,27 @@ 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, + 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(()) } @@ -689,7 +780,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 = (); @@ -703,6 +794,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(()) } diff --git a/crates/protocol/src/metadata.rs b/crates/protocol/src/metadata.rs index d5989e83..0e702450 100644 --- a/crates/protocol/src/metadata.rs +++ b/crates/protocol/src/metadata.rs @@ -57,6 +57,9 @@ const CLAUDE_PARENT_AGENT_ID_HEADER: &str = "x-claude-code-parent-agent-id"; // OpenCode session header — used for session_id correlation only (not a routing signal). const OPENCODE_SESSION_ID_HEADER: &str = "x-session-id"; +// Harbor benchmark proxy session header. +const HARBOR_PROXY_SESSION_ID_HEADER: &str = "proxy_x_session_id"; + // Generic Codex-compatible correlation headers. const SESSION_ID_HEADER: &str = "session-id"; const THREAD_ID_HEADER: &str = "thread-id"; @@ -83,6 +86,7 @@ const HEADER_CONFIG: &HeaderConfig = &[ RELAY_SESSION_ID_HEADER, OPENCODE_SESSION_ID_HEADER, CODEX_SESSION_ID_PATH, + HARBOR_PROXY_SESSION_ID_HEADER, SESSION_ID_HEADER, ], ), @@ -463,6 +467,28 @@ mod tests { assert!(!root.is_subagent); } + #[test] + fn normalizes_harbor_proxy_session_id() { + let harbor = metadata(&[("proxy_x_session_id", "harbor-attempt")]); + assert_eq!(harbor.session_id.as_deref(), Some("harbor-attempt")); + + let explicit = metadata(&[ + ("session-id", "generic-session"), + ("proxy_x_session_id", "harbor-attempt"), + ("x-switchyard-session-id", "explicit-session"), + ]); + assert_eq!(explicit.session_id.as_deref(), Some("explicit-session")); + + let harbor_over_generic = metadata(&[ + ("session-id", "generic-session"), + ("proxy_x_session_id", "harbor-attempt"), + ]); + assert_eq!( + harbor_over_generic.session_id.as_deref(), + Some("harbor-attempt") + ); + } + #[test] fn normalizes_correlation_and_session_headers_without_routing() { // Integrating-host headers are correlation data, not routing signals. diff --git a/crates/switchyard-server/src/lib.rs b/crates/switchyard-server/src/lib.rs index 13374da5..6c923a14 100644 --- a/crates/switchyard-server/src/lib.rs +++ b/crates/switchyard-server/src/lib.rs @@ -722,7 +722,9 @@ impl RequestLogContext { requested_model = self.requested_model.as_deref().unwrap_or(""), selected_model, streaming = self.streaming, - session_id = self.session_id.as_deref().unwrap_or(""), + // Affinity keys on this field, so "the harness sent no session header" and + // "it sent an empty one" are different diagnoses and must not both log as "". + session_id = self.session_id.as_deref().unwrap_or("-"), correlation_id = self.correlation_id.as_deref().unwrap_or(""), handling_duration_ms = duration_ms, error,