switchyard_libsy/algorithms/util/

affinity.rs

// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
// SPDX-License-Identifier: Apache-2.0

//! Model affinity as a single SDK component.
//!
//! [`AffinityRouter`] retains the first model and category chosen for a request's stable identity
//! and forces that decision on later requests sharing the identity. It is one object that plays
//! both SDK roles, so registering it as a processor and a classifier cannot drift apart:
//!
//! - As a [`Processor`] it *writes* the assignment: [`Event::Decision`] carries the request
//!   whose chosen model is retained, with the first assignment for an identity winning.
//! - As a [`Classifier`] it *reads* the assignment: it scores the retained model with
//!   confidence `1.0`, or returns no scores to abstain.
//!
//! Identity is derived from correlation metadata: by default, a request is keyed by its
//! session, and a sub-agent is keyed more finely by `session + agent` — a subset of its
//! session. [`AffinityRouter::for_subagents`] narrows affinity to delegated child-agent
//! work, leaving root and harness-maintenance traffic to later classifiers on every turn.

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;
use serde::Deserialize;
use switchyard_protocol::{Category, ContentBlock, Message, ModelId, Request, Role};

use crate::core::algorithm::{Driver, RoutingIdentity};
use crate::core::classifier::{Classification, Classifier, Score};
use crate::core::processor::{Event, Processor};
use crate::core::state::State;

/// Upper bound on retained assignments, keeping the process-local map from growing
/// without limit; the oldest entry is evicted once the bound is reached.
const MAX_ASSIGNMENTS: usize = 4096;

/// How often the classifier re-decides a session's target.
#[derive(Clone, Copy, Debug, Default, Deserialize, PartialEq, Eq)]
#[serde(rename_all = "snake_case")]
pub enum ClassifyTrigger {
    /// Judge every request, tool continuations included.
    #[default]
    EveryRequest,
    /// Judge each new user message, holding that target across the tool calls between.
    UserTurn,
    /// Judge once and reuse that target for the session.
    NewSession,
}

/// Tool results and standalone Codex environment updates do not start a user turn.
pub(crate) fn is_user_turn(message: &Message) -> bool {
    message.role == Role::User
        && message.content.iter().any(|block| {
            !matches!(block, ContentBlock::ToolResult(_)) && !is_environment_context(block)
        })
}

fn is_environment_context(block: &ContentBlock) -> bool {
    matches!(block, ContentBlock::Text { text } if text.trim()
        .strip_prefix("<environment_context>")
        .and_then(|context| context.split_once("</environment_context>"))
        .is_some_and(|(_, rest)| rest.trim().is_empty()))
}

fn is_environment_update(message: &Message) -> bool {
    message.role == Role::User
        && !message.content.is_empty()
        && message.content.iter().all(is_environment_context)
}

/// Inline instructions and standalone Codex environment updates sit between turns.
/// Turn detection skips them when it looks for the previous message.
pub(crate) fn is_turn_context(message: &Message) -> bool {
    matches!(message.role, Role::System | Role::Developer) || is_environment_update(message)
}

pub(crate) fn has_new_user_turn(messages: &[Message]) -> bool {
    messages
        .iter()
        .rev()
        .find(|message| !is_turn_context(message))
        .is_some_and(is_user_turn)
}

/// Who started the request, `user` or `agent` (GitHub Copilot's convention).
const INITIATOR_HEADER: &str = "x-initiator";

/// The client's `x-initiator` header: `Some(true)` when a person wrote the
/// request, `Some(false)` for the agent's own follow-up calls, `None` when the
/// header is absent or unrecognized and the message shape decides.
pub(crate) fn initiated_by_user(request: &Request) -> Option<bool> {
    let value = request
        .metadata
        .as_ref()?
        .http_headers
        .as_ref()?
        .get(INITIATOR_HEADER)?
        .to_str()
        .ok()?
        .trim();
    if value.eq_ignore_ascii_case("user") {
        Some(true)
    } else if value.eq_ignore_ascii_case("agent") {
        Some(false)
    } else {
        None
    }
}

/// Whether `request` opens a user turn: the `x-initiator` header says so, or else
/// its newest message is the user's.
pub(crate) fn starts_user_turn(request: &Request) -> bool {
    initiated_by_user(request).unwrap_or_else(|| has_new_user_turn(&request.llm_request.messages))
}

/// Retains a model per request identity and forces it on later matching requests.
///
/// Register the same instance as both a processor and a classifier; the two roles share
/// the retained assignments through this instance's interior storage. Decision events carry
/// their originating request, so concurrent turns cannot bind one request's identity to
/// another request's selected model.
///
/// [`with_latch_only`](Self::with_latch_only) narrows *which* models are retained — a
/// decision for any other model routes normally but is not latched (the escalation latch:
/// retain only the strong tier, never the weak one).
#[derive(Default)]
pub struct AffinityRouter {
    /// When set, only these models are retained; a decision for any other model is not latched.
    latch_only: Option<HashSet<ModelId>>,
    /// Whether requests other than delegated sub-agent work should abstain.
    subagents_only: bool,
    /// In absence of headers, use the message hash based fallback key to do task based routing
    message_hash_fallback: bool,
    /// Whether a new user message drops the assignment so the judge runs again.
    release_on_user_turn: bool,
    /// Retained assignments, shared across this router's processor and classifier roles.
    ///
    /// 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<RoutingIdentity, (ModelId, Option<Category>)>>,
    /// Whether the "no identity to key on" warning has already been emitted.
    unkeyed_warning_emitted: AtomicBool,
}

impl AffinityRouter {
    /// Creates a router that latches every decision.
    pub fn new() -> Self {
        Self::default()
    }

    /// Creates a router that retains assignments only for delegated child-agent work.
    ///
    /// Root and harness-maintenance requests always abstain, so a later classifier selects
    /// them on every turn.
    pub fn for_subagents() -> Self {
        Self {
            subagents_only: true,
            ..Self::default()
        }
    }

    /// Uses the first user-message text as a fallback key when metadata has no session.
    pub fn with_message_hash_fallback(mut self) -> Self {
        self.message_hash_fallback = true;
        self
    }

    /// Re-decides per user turn, holding the target across the tool calls between.
    pub fn with_release_on_user_turn(mut self) -> Self {
        self.release_on_user_turn = true;
        self
    }

    /// Restricts latching to `models`; a decision for any other model routes but is not
    /// retained.
    pub fn with_latch_only(mut self, models: impl IntoIterator<Item = impl Into<ModelId>>) -> Self {
        self.latch_only = Some(models.into_iter().map(Into::into).collect());
        self
    }

    /// Whether a decision for `model` should be retained.
    fn should_latch(&self, model: &str) -> bool {
        self.latch_only
            .as_ref()
            .is_none_or(|set| set.contains(model))
    }

    /// Derives the stable identity this router should retain for `request`.
    fn affinity_key(&self, request: &Request) -> Option<RoutingIdentity> {
        let metadata = request.metadata.as_ref();
        let is_subagent = metadata.is_some_and(|metadata| metadata.is_subagent);
        let is_subagent_work = metadata.is_some_and(|metadata| metadata.is_subagent_work());
        // This mode handles only delegated work; root and maintenance requests fall through.
        if self.subagents_only && !is_subagent_work {
            return None;
        }

        // Child requests require their explicit session + agent identity and never fall
        // back to task text.
        let key = retention_key(request, self.message_hash_fallback && !is_subagent);
        // 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)
    }
}

#[async_trait]
impl Processor for AffinityRouter {
    async fn process(&self, _state: &mut State, event: Event<'_>) -> crate::Result<()> {
        if let Event::Decision {
            request,
            selected_model_id,
            category,
            ..
        } = event
            && let Some(key) = self.affinity_key(request)
        {
            let mut assignments = self.assignments.lock();
            let writable = self.release_on_user_turn || !assignments.contains_key(&key);
            if self.should_latch(selected_model_id) && writable {
                evict_if_full(&mut assignments);
                assignments.insert(key, (selected_model_id.clone(), category));
            } else if self.release_on_user_turn && starts_user_turn(request) {
                // The new turn was decided for a model that is not retained, so the
                // previous turn's hold must not reach this turn's tool calls.
                assignments.remove(&key);
            }
        }
        Ok(())
    }
}

/// The key a request's retained value is stored under.
///
/// Prefers the request's own identity and falls back to hashing the first user
/// message when the caller allows it, which is the only handle on a client that
/// sends no session ID.
pub(crate) fn retention_key(request: &Request, hash_fallback: bool) -> Option<RoutingIdentity> {
    if let Some(identity) = RoutingIdentity::from_request(request) {
        return Some(identity);
    }
    if !hash_fallback {
        return None;
    }
    first_user_message_hash(request).map(|hash| {
        tracing::debug!(retention_key = %hash, "retaining by message hash fallback");
        RoutingIdentity::Session(hash)
    })
}

/// Hashes the first user message so later turns retain the initial task's affinity.
/// For benchmarking purpose with harnesses, task instructions are added as a user prompt to the request so we hash the initial user message.
/// TODO: Have not considered multi-modal payloads yet. That needs to be handled separately.
fn first_user_message_hash(request: &Request) -> Option<String> {
    let message = request
        .llm_request
        .messages
        .iter()
        .find(|message| message.role == Role::User)?;
    let mut hasher = DefaultHasher::new();
    message.text_content("")?.hash(&mut hasher);
    Some(format!("{:016x}", hasher.finish()))
}

#[async_trait]
impl Classifier for AffinityRouter {
    async fn score(
        &self,
        _state: &mut State,
        request: &mut Request,
        driver: &Driver,
    ) -> crate::Result<(Classification, Option<switchyard_protocol::Response>)> {
        let Some(key) = self.affinity_key(request) else {
            return Ok((Classification::Scores(Vec::new()), None));
        };
        // Abstaining rather than clearing keeps this read path free of writes, so the judge
        // decides the turn whatever order concurrent latches land in.
        if self.release_on_user_turn && starts_user_turn(request) {
            return Ok((Classification::Scores(Vec::new()), None));
        }
        // An empty `any` group carries no information about which models are still
        // available, so it must not be read as "every assignment is now stale".
        let available = driver.models_for(&Category::Any);
        let mut assignments = self.assignments.lock();
        let assigned = assignments.get(&key).cloned();
        let assigned = match assigned.as_ref() {
            Some((target, _)) if !available.is_empty() && !available.contains(target) => {
                assignments.remove(&key);
                None
            }
            assigned => assigned,
        };
        if assigned.is_some() {
            driver.set_evidence(serde_json::json!({"source": "retained"}));
        }
        Ok((
            Classification::Scores(match assigned {
                Some((target, category)) => vec![Score {
                    confidence: 1.0,
                    target: target.clone(),
                    category: category.clone(),
                }],
                None => Vec::new(),
            }),
            None,
        ))
    }
}

/// Evicts one arbitrary assignment when the map has reached [`MAX_ASSIGNMENTS`].
pub(crate) fn evict_if_full<V>(retained: &mut HashMap<RoutingIdentity, V>) {
    if retained.len() >= MAX_ASSIGNMENTS
        && let Some(evicted) = retained.keys().next().cloned()
    {
        retained.remove(&evicted);
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    use std::sync::Arc;

    use switchyard_protocol::{LlmRequest, Metadata, ToolResult, text_request};

    use crate::core::algorithm::RuntimeModels;

    /// Boxed, thread-safe error type keeping the test helpers ergonomic.
    type BoxErr = Box<dyn std::error::Error + Send + Sync>;

    #[test]
    fn environment_updates_preserve_real_user_content() {
        let context =
            "<environment_context><current_date>2026-10-31</current_date></environment_context>";
        assert!(!is_user_turn(&Message::text(
            Role::User,
            format!(" {context}\n")
        )));
        assert!(is_user_turn(&Message::text(
            Role::User,
            format!("{context}\nfix the parser")
        )));
        assert!(is_user_turn(&Message::text(
            Role::User,
            format!("explain this: {context}")
        )));
        let mut mixed = Message::text(Role::User, context);
        mixed.content.push(ContentBlock::Text {
            text: "fix the parser".into(),
        });
        assert!(is_user_turn(&mixed));
        for (role, expected) in [
            (Role::User, true),
            (Role::Assistant, false),
            (Role::Tool, false),
        ] {
            assert_eq!(
                has_new_user_turn(&[
                    Message::text(role, "follow-up"),
                    Message::text(Role::User, context),
                    Message::text(Role::System, "System guidance"),
                    Message::text(Role::Developer, "Developer guidance"),
                ]),
                expected
            );
        }
    }

    fn fixed_model(target: &str) -> ModelId {
        ModelId::from(target)
    }

    fn driver() -> Driver {
        Driver::new(
            "test",
            Arc::new(RuntimeModels::new(
                [(
                    Category::Any,
                    ["model-a", "model-b", "weak", "strong"]
                        .map(ModelId::from)
                        .to_vec(),
                )]
                .into(),
            )),
        )
        .0
    }

    fn request(metadata: Metadata) -> Request {
        Request {
            llm_request: text_request(Some("auto".to_string()), "hi"),
            raw_request: None,
            metadata: Some(metadata),
        }
    }

    fn task_request(
        metadata: Option<Metadata>,
        first_user: &str,
        follow_up: Option<&str>,
    ) -> Request {
        let mut messages = vec![
            Message::text(Role::System, "follow repository instructions"),
            Message::text(Role::User, first_user),
        ];
        if let Some(follow_up) = follow_up {
            messages.push(Message::text(Role::Assistant, "I will inspect the code."));
            messages.push(Message::text(Role::User, follow_up));
        }
        Request {
            llm_request: LlmRequest {
                model: Some("auto".to_string()),
                messages,
                ..LlmRequest::default()
            },
            raw_request: None,
            metadata,
        }
    }

    fn session(session_id: &str, agent_id: &str) -> Metadata {
        Metadata {
            session_id: Some(session_id.to_string()),
            agent_id: Some(agent_id.to_string()),
            ..Metadata::default()
        }
    }

    fn subagent(agent_id: &str, task_id: &str) -> Metadata {
        Metadata {
            session_id: Some("session-1".to_string()),
            agent_id: Some(agent_id.to_string()),
            task_id: Some(task_id.to_string()),
            is_subagent: true,
            is_delegated_work: true,
            ..Metadata::default()
        }
    }

    /// Folds a request and its decision through the router, retaining `model`.
    async fn retain(
        router: &AffinityRouter,
        state: &mut State,
        request: &mut Request,
        model: &'static str,
    ) -> Result<(), BoxErr> {
        let selected_model_id = ModelId::from(model);
        router
            .process(
                state,
                Event::Decision {
                    request,
                    selected_model_id: &selected_model_id,
                    category: None,
                    driver: &driver(),
                },
            )
            .await?;
        Ok(())
    }

    /// Scores through the definitive classification variant used by affinity.
    async fn scores(
        classifier: &dyn Classifier,
        state: &mut State,
        request: &mut Request,
    ) -> Result<Vec<Score>, BoxErr> {
        match classifier.score(state, request, &driver()).await?.0 {
            Classification::Scores(scores) => Ok(scores),
            Classification::Ambiguous(_) => Err("affinity never returns ambiguous scores".into()),
        }
    }

    #[tokio::test]
    async fn session_retains_first_model_across_requests() -> Result<(), BoxErr> {
        let router = AffinityRouter::new();
        let mut state = State::default();

        let mut first = request(session("session-1", "agent-a"));
        retain(&router, &mut state, &mut first, "model-a").await?;

        // A different agent in the same session is scored onto the retained model.
        let mut second = request(session("session-1", "agent-b"));
        let scores = scores(&router, &mut state, &mut second).await?;
        assert_eq!(scores.len(), 1);
        assert_eq!(scores[0].confidence, 1.0);
        assert_eq!(scores[0].target, "model-a");
        Ok(())
    }

    #[tokio::test]
    async fn subagent_only_retains_children_without_latching_root_traffic() -> Result<(), BoxErr> {
        let router = AffinityRouter::for_subagents();
        let mut state = State::default();

        let mut root = request(session("session-1", "root-agent"));
        retain(&router, &mut state, &mut root, "model-a").await?;
        assert!(scores(&router, &mut state, &mut root).await?.is_empty());

        let mut first_child_turn = request(subagent("child-1", "task-1"));
        retain(&router, &mut state, &mut first_child_turn, "model-b").await?;
        let mut later_child_turn = request(subagent("child-1", "task-2"));
        let scores = scores(&router, &mut state, &mut later_child_turn).await?;
        assert_eq!(
            scores.first().map(|score| score.target.as_str()),
            Some("model-b")
        );
        Ok(())
    }

    #[tokio::test]
    async fn first_decision_wins() -> Result<(), BoxErr> {
        let router = AffinityRouter::new();
        let mut state = State::default();

        let mut req = request(session("session-1", "agent-a"));
        retain(&router, &mut state, &mut req, "model-a").await?;
        // A later decision for the same identity must not overwrite the first.
        retain(&router, &mut state, &mut req, "model-b").await?;

        let scores = scores(&router, &mut state, &mut req).await?;
        assert_eq!(
            scores.first().map(|score| score.target.as_str()),
            Some("model-a")
        );
        Ok(())
    }

    #[tokio::test]
    async fn subagent_is_keyed_by_agent_not_task() -> Result<(), BoxErr> {
        let router = AffinityRouter::new();
        let mut state = State::default();

        let mut first = request(subagent("child-1", "task-1"));
        retain(&router, &mut state, &mut first, "model-a").await?;

        // Same child, different task: still scored onto the retained model.
        let mut second = request(subagent("child-1", "task-2"));
        let scores = scores(&router, &mut state, &mut second).await?;
        assert_eq!(
            scores.first().map(|score| score.target.as_str()),
            Some("model-a")
        );
        Ok(())
    }

    #[tokio::test]
    async fn distinct_subagents_are_assigned_independently() -> Result<(), BoxErr> {
        let router = AffinityRouter::new();
        let mut state = State::default();

        // One child in the session is pinned...
        retain(
            &router,
            &mut state,
            &mut request(subagent("child-1", "task-1")),
            "model-a",
        )
        .await?;

        // ...a sibling child in the same session has no assignment of its own yet.
        let mut sibling = request(subagent("child-2", "task-1"));
        assert!(scores(&router, &mut state, &mut sibling).await?.is_empty());
        Ok(())
    }

    #[tokio::test]
    async fn subagent_does_not_inherit_session_assignment() -> Result<(), BoxErr> {
        let router = AffinityRouter::new();
        let mut state = State::default();

        // The session root is pinned, but a sub-agent is keyed separately...
        retain(
            &router,
            &mut state,
            &mut request(session("session-1", "root-1")),
            "model-a",
        )
        .await?;

        // ...so the sub-agent abstains until it is assigned in its own right.
        let mut child = request(subagent("child-1", "task-1"));
        assert!(scores(&router, &mut state, &mut child).await?.is_empty());
        Ok(())
    }

    #[tokio::test]
    async fn classifier_abstains_without_a_session() -> Result<(), BoxErr> {
        let router = AffinityRouter::new();
        let mut state = State::default();

        // No session id at all: nothing to key on.
        let mut req = request(Metadata::default());
        assert!(scores(&router, &mut state, &mut req).await?.is_empty());
        Ok(())
    }

    #[tokio::test]
    async fn message_hash_fallback_uses_the_first_user_message() -> Result<(), BoxErr> {
        let router = AffinityRouter::new().with_message_hash_fallback();
        let mut state = State::default();

        let mut first = task_request(
            None,
            "Add a unit test for this function.",
            Some("Now run the test suite."),
        );
        retain(&router, &mut state, &mut first, "weak").await?;

        let mut follow_up = task_request(
            None,
            "Add a unit test for this function.",
            Some("Now file a pull request."),
        );
        assert_eq!(
            scores(&router, &mut state, &mut follow_up)
                .await?
                .first()
                .map(|score| score.target.as_str()),
            Some("weak")
        );

        let mut other_task = task_request(
            None,
            "Reimplement this binary from two input/output pairs.",
            Some("Now run the test suite."),
        );
        assert!(
            scores(&router, &mut state, &mut other_task)
                .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 = State::default();

        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 {
            llm_request: LlmRequest {
                messages: vec![user_message],
                ..LlmRequest::default()
            },
            raw_request: None,
            metadata: None,
        };
        let text_only = request(Message::text(Role::User, "Implement the parser."));
        let text_with_reasoning = request(Message {
            role: Role::User,
            content: vec![
                ContentBlock::Text {
                    text: "Implement the parser.".to_string(),
                },
                ContentBlock::Reasoning {
                    text: "Internal provider reasoning.".to_string(),
                    signature: Some("provider-signature".to_string()),
                    details: Vec::new(),
                },
            ],
        });

        assert_eq!(
            first_user_message_hash(&text_only),
            first_user_message_hash(&text_with_reasoning)
        );
    }

    #[tokio::test]
    async fn metadata_session_takes_precedence_over_message_hash() -> Result<(), BoxErr> {
        let router = AffinityRouter::new().with_message_hash_fallback();
        let mut state = State::default();

        let mut first = task_request(
            Some(session("session-1", "agent-a")),
            "Implement the parser.",
            None,
        );
        retain(&router, &mut state, &mut first, "strong").await?;

        let mut other_session = task_request(
            Some(session("session-2", "agent-a")),
            "Implement the parser.",
            None,
        );
        assert!(
            scores(&router, &mut state, &mut other_session)
                .await?
                .is_empty()
        );
        Ok(())
    }

    #[tokio::test]
    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 = State::default();
            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(())
    }

    #[tokio::test]
    async fn one_router_serves_both_roles() -> Result<(), BoxErr> {
        // The same instance is registered under both SDK roles; a decision folded in via
        // the processor handle is read back via the classifier handle.
        let router = Arc::new(AffinityRouter::new());
        let processor: Arc<dyn Processor> = router.clone();
        let classifier: Arc<dyn Classifier> = router;
        let mut state = State::default();

        let mut first = request(session("session-1", "agent-a"));
        processor
            .process(
                &mut state,
                Event::Decision {
                    request: &mut first,
                    selected_model_id: &fixed_model("model-a"),
                    category: None,
                    driver: &driver(),
                },
            )
            .await?;

        let mut second = request(session("session-1", "agent-b"));
        let scores = scores(classifier.as_ref(), &mut state, &mut second).await?;
        assert_eq!(
            scores.first().map(|score| score.target.as_str()),
            Some("model-a")
        );
        Ok(())
    }

    #[tokio::test]
    async fn decision_without_an_affinity_identity_is_ignored() -> Result<(), BoxErr> {
        let router = AffinityRouter::new();
        let mut state = State::default();
        let mut unkeyed = request(Metadata::default());

        router
            .process(
                &mut state,
                Event::Decision {
                    request: &mut unkeyed,
                    selected_model_id: &fixed_model("model-a"),
                    category: None,
                    driver: &driver(),
                },
            )
            .await?;

        let mut req = request(session("session-1", "agent-a"));
        assert!(scores(&router, &mut state, &mut req).await?.is_empty());
        Ok(())
    }

    #[tokio::test]
    async fn decisions_retain_their_originating_request_identity() -> Result<(), BoxErr> {
        let router = AffinityRouter::new();
        let mut state = State::default();
        let mut first = request(session("session-1", "agent-a"));
        let mut second = request(session("session-2", "agent-b"));

        // Replay decisions in the opposite order. Each decision carries its request,
        // so the assignments cannot cross.
        router
            .process(
                &mut state,
                Event::Decision {
                    request: &mut second,
                    selected_model_id: &fixed_model("model-b"),
                    category: None,
                    driver: &driver(),
                },
            )
            .await?;
        router
            .process(
                &mut state,
                Event::Decision {
                    request: &mut first,
                    selected_model_id: &fixed_model("model-a"),
                    category: None,
                    driver: &driver(),
                },
            )
            .await?;

        let first_scores = scores(&router, &mut state, &mut first).await?;
        let second_scores = scores(&router, &mut state, &mut second).await?;
        assert_eq!(
            first_scores.first().map(|score| score.target.as_str()),
            Some("model-a")
        );
        assert_eq!(
            second_scores.first().map(|score| score.target.as_str()),
            Some("model-b")
        );
        Ok(())
    }

    #[tokio::test]
    async fn distinct_sessions_are_assigned_independently() -> Result<(), BoxErr> {
        let router = AffinityRouter::new();
        let mut state = State::default();

        retain(
            &router,
            &mut state,
            &mut request(session("session-1", "agent-a")),
            "model-a",
        )
        .await?;
        retain(
            &router,
            &mut state,
            &mut request(session("session-2", "agent-a")),
            "model-b",
        )
        .await?;

        let first = scores(
            &router,
            &mut state,
            &mut request(session("session-1", "other")),
        )
        .await?;
        let second = scores(
            &router,
            &mut state,
            &mut request(session("session-2", "other")),
        )
        .await?;
        assert_eq!(
            first.first().map(|score| score.target.as_str()),
            Some("model-a")
        );
        assert_eq!(
            second.first().map(|score| score.target.as_str()),
            Some("model-b")
        );
        Ok(())
    }

    #[tokio::test]
    async fn subagent_without_an_agent_id_abstains_and_warns() -> Result<(), BoxErr> {
        let router = AffinityRouter::new();
        let mut state = State::default();

        // The sub-agent flag is set but no agent id is present, so no key can be formed;
        // the request is neither retained nor scored.
        let metadata = Metadata {
            session_id: Some("session-1".to_string()),
            is_subagent: true,
            ..Metadata::default()
        };
        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(())
    }

    #[tokio::test]
    async fn assignments_are_bounded_by_the_cap() -> Result<(), BoxErr> {
        let router = AffinityRouter::new();
        let mut state = State::default();

        // One distinct session past the cap forces exactly one eviction.
        for index in 0..=MAX_ASSIGNMENTS {
            let session_id = format!("session-{index}");
            retain(
                &router,
                &mut state,
                &mut request(session(&session_id, "agent-a")),
                "model-a",
            )
            .await?;
        }

        let len = router.assignments.lock().len();
        assert_eq!(len, MAX_ASSIGNMENTS);
        Ok(())
    }

    #[tokio::test]
    async fn release_on_user_turn_drops_the_assignment_only_when_the_user_speaks()
    -> Result<(), BoxErr> {
        let router = AffinityRouter::new().with_release_on_user_turn();
        let mut state = State::default();
        let mut opening = task_request(Some(session("session-1", "agent-a")), "add caching", None);
        retain(&router, &mut state, &mut opening, "weak").await?;

        // A tool continuation holds the assignment, so no judge call.
        let mut continued = opening.clone();
        continued.llm_request.messages.push(Message {
            role: Role::User,
            content: vec![ContentBlock::ToolResult(ToolResult {
                tool_call_id: "call-1".to_string(),
                content: Vec::new(),
                is_error: None,
            })],
        });
        assert_eq!(
            scores(&router, &mut state, &mut continued)
                .await?
                .first()
                .map(|s| s.target.as_str()),
            Some("weak")
        );

        // A new user message releases it, so the turn abstains and the judge runs.
        let mut spoke = task_request(
            Some(session("session-1", "agent-a")),
            "add caching",
            Some("no, shared across processes"),
        );
        assert!(scores(&router, &mut state, &mut spoke).await?.is_empty());
        Ok(())
    }

    #[tokio::test]
    async fn a_user_turn_decided_for_an_unlatched_model_ends_the_previous_hold()
    -> Result<(), BoxErr> {
        let router = AffinityRouter::new()
            .with_release_on_user_turn()
            .with_latch_only(["strong"]);
        let mut state = State::default();
        let mut escalated =
            task_request(Some(session("session-1", "agent-a")), "add caching", None);
        retain(&router, &mut state, &mut escalated, "strong").await?;

        let mut spoke = task_request(
            Some(session("session-1", "agent-a")),
            "add caching",
            Some("now add a test"),
        );
        retain(&router, &mut state, &mut spoke, "weak").await?;

        // The new turn's tool continuation is not held on the earlier turn's model.
        let mut continued = spoke.clone();
        continued.llm_request.messages.push(Message {
            role: Role::User,
            content: vec![ContentBlock::ToolResult(ToolResult {
                tool_call_id: "call-1".to_string(),
                content: Vec::new(),
                is_error: None,
            })],
        });
        assert!(
            scores(&router, &mut state, &mut continued)
                .await?
                .is_empty()
        );
        Ok(())
    }

    #[tokio::test]
    async fn latch_only_retains_matching_models() -> Result<(), BoxErr> {
        let router = AffinityRouter::new().with_latch_only(["strong"]);
        let mut state = State::default();
        let mut req = request(session("session-1", "agent-a"));

        // A "weak" decision is not retained — a later turn is not latched.
        retain(&router, &mut state, &mut req, "weak").await?;
        assert!(scores(&router, &mut state, &mut req).await?.is_empty());

        // A "strong" decision is retained — later turns latch onto it.
        retain(&router, &mut state, &mut req, "strong").await?;
        assert_eq!(
            scores(&router, &mut state, &mut req)
                .await?
                .first()
                .map(|s| s.target.as_str()),
            Some("strong")
        );
        Ok(())
    }

    #[tokio::test]
    async fn an_unprovisioned_any_group_does_not_evict_assignments() -> Result<(), BoxErr> {
        let router = AffinityRouter::new();
        let mut state = State::default();
        let mut req = request(session("session-1", "agent-a"));
        retain(&router, &mut state, &mut req, "model-a").await?;

        let empty = Driver::new("test", Arc::new(RuntimeModels::default())).0;
        let (classification, _) = router.score(&mut state, &mut req, &empty).await?;
        let Classification::Scores(retained) = classification else {
            return Err("affinity never returns ambiguous scores".into());
        };
        assert_eq!(retained.first().map(|s| s.target.as_str()), Some("model-a"));

        // The assignment survived, so a later turn with the group present still latches.
        assert_eq!(
            scores(&router, &mut state, &mut req)
                .await?
                .first()
                .map(|s| s.target.as_str()),
            Some("model-a")
        );
        Ok(())
    }
}