fix: harden session affinity identity handling

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<HashMap<AffinityKey, String>>,
+    /// 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<AffinityKey> {
-        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(())
     }
 

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.

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,