switchyard_libsy/algorithms/
composite.rs

1// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
2// SPDX-License-Identifier: Apache-2.0
3
4//! Routing that stacks a judge over a stage router.
5//!
6//! The judge runs as a [`Processor`]: it sets configuration the stage router reads,
7//! and picks no target itself.
8
9use std::collections::HashMap;
10use std::sync::Arc;
11
12use async_trait::async_trait;
13use parking_lot::Mutex;
14
15use super::fall_through::FallThrough;
16use super::llm_class::{
17    CustomClassifierConfig, LlmClassifierConfig, LlmTaskClassifier, TaskClassifierConfig,
18};
19use super::stage::{StageRouterConfig, build_stage_route};
20use super::util::affinity::{ClassifyTrigger, evict_if_full, has_new_user_turn, retention_key};
21use super::util::stage::{PickerMode, Tier, fall_open_tier, set_fall_open};
22use crate::core::algorithm::{Algorithm, Driver, RoutingIdentity};
23use crate::core::classifier::{Classification, Classifier};
24use crate::core::processor::{Event, Processor};
25use crate::core::state::State;
26use crate::{LibsyError, Result};
27use switchyard_protocol::{Category, Request, Response};
28
29const COMPOSITE: &str = "composite";
30
31/// Sets the stage router's fall-open tier from a judge verdict.
32///
33/// Retains the tier per routing identity, so it survives requests that carry no
34/// session ID when `message_hash_fallback` is on. The retained tier is replayed
35/// into state on every request so the cascade below reads it.
36struct TierSetter {
37    judge: Arc<dyn Classifier>,
38    trigger: ClassifyTrigger,
39    message_hash_fallback: bool,
40    default_tier: Tier,
41    tiers: Mutex<HashMap<RoutingIdentity, Tier>>,
42}
43
44impl TierSetter {
45    fn tier_for(category: Option<Category>) -> Option<Tier> {
46        match category {
47            Some(Category::Capable) => Some(Tier::Capable),
48            Some(Category::Efficient) => Some(Tier::Efficient),
49            _ => None,
50        }
51    }
52
53    /// Two requests for one identity can both pass this and both judge, since a
54    /// judge call sits between here and [`retain`](Self::retain). The later wins.
55    fn is_due(&self, identity: Option<&RoutingIdentity>, state: &State, request: &Request) -> bool {
56        match self.trigger {
57            ClassifyTrigger::UserTurn => has_new_user_turn(&request.llm_request.messages),
58            // Unkeyed requests with no carried tier cannot be told apart, so every one
59            // is a new session.
60            ClassifyTrigger::NewSession => {
61                identity.is_none_or(|identity| !self.tiers.lock().contains_key(identity))
62                    && fall_open_tier(state).is_none()
63            }
64            // Rejected by the constructor, and only in the enum for the standalone route.
65            ClassifyTrigger::EveryRequest => true,
66        }
67    }
68
69    fn retain(&self, identity: RoutingIdentity, tier: Tier) {
70        let mut tiers = self.tiers.lock();
71        // A user turn re-decides, so it overwrites. A session keeps its first verdict,
72        // matching how affinity retains an assignment.
73        let writable = self.trigger == ClassifyTrigger::UserTurn || !tiers.contains_key(&identity);
74        if writable {
75            evict_if_full(&mut tiers);
76            tiers.insert(identity, tier);
77        }
78    }
79}
80
81#[async_trait]
82impl Processor for TierSetter {
83    async fn process(&self, state: &mut State, event: Event<'_>) -> Result<()> {
84        let Event::Request { request, driver } = event else {
85            return Ok(());
86        };
87        let identity = retention_key(request, self.message_hash_fallback);
88        if self.is_due(identity.as_ref(), state, request) {
89            let (classification, _) = self.judge.score(state, request, driver).await?;
90            if let Some(winner) = classification.argmax(false)?
91                && let Some(tier) = Self::tier_for(winner.category)
92            {
93                set_fall_open(state, tier);
94                if let Some(identity) = identity {
95                    self.retain(identity, tier);
96                }
97                return Ok(());
98            }
99            if self.trigger == ClassifyTrigger::NewSession
100                && let Some(identity) = identity.as_ref()
101            {
102                // A failed opening judgment is still the session's one judgment.
103                self.retain(identity.clone(), self.default_tier);
104            }
105        }
106        // Either not this request's turn or the judge had no verdict, so the last
107        // tier stands rather than dropping back to the picker default.
108        if let Some(tier) = identity
109            .and_then(|identity| self.tiers.lock().get(&identity).copied())
110            .or_else(|| fall_open_tier(state))
111        {
112            set_fall_open(state, tier);
113            driver.set_evidence_if_empty(serde_json::json!({"source": "retained"}));
114        }
115        Ok(())
116    }
117}
118
119/// Classifier settings for the composite router's retained tier.
120pub enum CompositeJudgeConfig {
121    /// Selects a tier from a capability verdict.
122    Capability(TaskClassifierConfig),
123    /// Selects a tier using a custom schema and policy, retaining the fallback on failure.
124    Custom(CustomClassifierConfig),
125}
126
127/// A judge stacked over a stage router.
128pub struct CompositeRouterConfig {
129    /// Judge settings, including how often `classify_trigger` runs it.
130    pub judge: CompositeJudgeConfig,
131    /// Serves the turns, with the tier the judge picked as its fall-open default.
132    pub stage: StageRouterConfig,
133}
134
135/// Runs a stage router with a tier the judge picks.
136pub struct CompositeRouter {
137    route: FallThrough,
138}
139
140impl CompositeRouter {
141    /// Stacks the judge over a stage router across the same tier pair.
142    ///
143    /// Errors on a configuration either algorithm rejects and on `every_request`.
144    ///
145    /// A stage router carrying its own judge is allowed, but that judge sits ahead
146    /// of the fall-open tier and so answers most of the turns this one set a tier for.
147    pub fn new(config: CompositeRouterConfig) -> Result<Self> {
148        let setter = Self::tier_setter(config.judge, config.stage.mode.default_tier())?;
149        let route = build_stage_route(config.stage)?
150            .with_name(COMPOSITE)
151            .with_processor(Arc::new(setter));
152        Ok(Self { route })
153    }
154
155    /// Builds two independent stage profiles selected by the judge.
156    pub fn new_with_profiles(
157        judge: CompositeJudgeConfig,
158        mut profile_0: StageRouterConfig,
159        mut profile_1: StageRouterConfig,
160    ) -> Result<Self> {
161        for profile in [&mut profile_0, &mut profile_1] {
162            if profile.mode == PickerMode::CapableFirst {
163                profile.efficient_hold_turns.get_or_insert(-1);
164            }
165        }
166        let setter = Self::tier_setter(judge, Tier::Efficient)?;
167        let profiles =
168            StageProfiles([build_stage_route(profile_0)?, build_stage_route(profile_1)?]);
169        let route = FallThrough::new_with_state()
170            .with_name(COMPOSITE)
171            .with_processor(Arc::new(setter))
172            .with_classifier(Arc::new(profiles));
173        Ok(Self { route })
174    }
175
176    fn tier_setter(judge: CompositeJudgeConfig, default_tier: Tier) -> Result<TierSetter> {
177        let (trigger, message_hash_fallback) = match &judge {
178            CompositeJudgeConfig::Capability(judge) => {
179                (judge.classify_trigger, judge.message_hash_fallback)
180            }
181            CompositeJudgeConfig::Custom(config) => {
182                (config.classify_trigger, config.message_hash_fallback)
183            }
184        };
185        if trigger == ClassifyTrigger::EveryRequest {
186            return Err(LibsyError::AlgorithmError {
187                message: "composite: classify_trigger must be user_turn or new_session".to_string(),
188            });
189        }
190        let judge: Arc<dyn Classifier> = match judge {
191            CompositeJudgeConfig::Capability(judge) => {
192                // Composite owns the trigger and retained verdict.
193                Arc::new(LlmTaskClassifier::new(LlmClassifierConfig::Capability {
194                    config: TaskClassifierConfig {
195                        classify_trigger: ClassifyTrigger::EveryRequest,
196                        message_hash_fallback: false,
197                        ..judge
198                    },
199                })?)
200            }
201            CompositeJudgeConfig::Custom(config) => {
202                LlmTaskClassifier::custom_classifier(&config, true)?
203            }
204        };
205        Ok(TierSetter {
206            judge,
207            trigger,
208            message_hash_fallback,
209            default_tier,
210            tiers: Mutex::new(HashMap::new()),
211        })
212    }
213}
214
215struct StageProfiles([FallThrough; 2]);
216
217#[async_trait]
218impl Classifier for StageProfiles {
219    async fn score(
220        &self,
221        state: &mut State,
222        request: &mut Request,
223        driver: &Driver,
224    ) -> Result<(Classification, Option<Response>)> {
225        let (profile, prefix) = match fall_open_tier(state) {
226            Some(Tier::Capable) => (&self.0[1], "composite.profile_1."),
227            _ => (&self.0[0], "composite.profile_0."),
228        };
229        // Each profile retains its own holds, including in caller-carried state.
230        let mut profile_state = State {
231            extra: state
232                .extra
233                .iter()
234                .filter_map(|(key, value)| {
235                    key.strip_prefix(prefix)
236                        .map(|key| (key.to_string(), value.clone()))
237                })
238                .collect(),
239            ..State::default()
240        };
241        let result = profile.route(&mut profile_state, driver, request).await;
242        state.extra.retain(|key, _| !key.starts_with(prefix));
243        state.extra.extend(
244            profile_state
245                .extra
246                .into_iter()
247                .map(|(key, value)| (format!("{prefix}{key}"), value)),
248        );
249        state.tool_signals = profile_state.tool_signals;
250        state.target_prompts = profile_state.target_prompts;
251        let (score, response) = result?;
252        Ok((Classification::Scores(vec![score]), response))
253    }
254}
255
256#[async_trait]
257impl Algorithm for CompositeRouter {
258    fn name(&self) -> &str {
259        COMPOSITE
260    }
261
262    async fn route(
263        self: Arc<Self>,
264        driver: Driver,
265        request: Request,
266    ) -> Result<crate::RoutingOutcome> {
267        self.route.execute(driver, request).await
268    }
269}
270
271#[cfg(test)]
272mod tests {
273    use crate::{CapabilityJudgeConfig, LlmCapabilityConfig};
274    use std::collections::HashMap;
275    use std::sync::Arc;
276
277    use switchyard_protocol::{Category, LlmClientError, Message, ModelId, Role};
278
279    use super::*;
280    use crate::algorithms::util::tier_fixtures::{JUDGE, Recorder, turn_request};
281    use crate::core::testing::test_drive_with_models;
282
283    fn runtime_models() -> HashMap<Category, Vec<ModelId>> {
284        [
285            (Category::Judge, vec![ModelId::from(JUDGE)]),
286            (Category::Efficient, vec![ModelId::from("weak")]),
287            (Category::Capable, vec![ModelId::from("strong")]),
288            (
289                Category::Any,
290                vec![ModelId::from("strong"), ModelId::from("weak")],
291            ),
292        ]
293        .into()
294    }
295
296    fn user_turn_request() -> Request {
297        let mut request = turn_request(false);
298        request
299            .llm_request
300            .messages
301            .push(Message::text(Role::User, "now rewrite the parser"));
302        request
303    }
304
305    /// The same request shape with no session ID, so only the hash can key it.
306    fn unkeyed(mut request: Request) -> Request {
307        if let Some(metadata) = request.metadata.as_mut() {
308            metadata.session_id = None;
309        }
310        request
311    }
312
313    fn hash_keyed_router() -> Result<Arc<CompositeRouter>> {
314        Ok(Arc::new(CompositeRouter::new(CompositeRouterConfig {
315            judge: CompositeJudgeConfig::Capability(TaskClassifierConfig {
316                judge: CapabilityJudgeConfig::Llm(LlmCapabilityConfig {
317                    base_threshold: 0.5,
318                    ..LlmCapabilityConfig::default()
319                }),
320                classify_trigger: ClassifyTrigger::UserTurn,
321                message_hash_fallback: true,
322                ..Default::default()
323            }),
324            stage: StageRouterConfig::new(PickerMode::EfficientFirst, 0.5),
325        })?))
326    }
327
328    fn router() -> Result<Arc<CompositeRouter>> {
329        router_with_trigger(ClassifyTrigger::UserTurn)
330    }
331
332    #[tokio::test]
333    async fn capable_first_defaults_to_pinning_efficient_after_a_switch() -> Result<()> {
334        for capable_first in [false, true] {
335            for turns in [None, Some(0), Some(1), Some(-1)] {
336                let mut stage = StageRouterConfig::new(PickerMode::EfficientFirst, 0.5);
337                stage.efficient_hold_turns = turns;
338                let router = router_with_stage(ClassifyTrigger::UserTurn, stage)?;
339                let recorder = Arc::new(Recorder::default());
340                *recorder.judge_p_solve.lock() = if capable_first { 0.0 } else { 1.0 };
341                let mut initial = turn_request(false);
342                initial.llm_request.messages.truncate(1);
343                let mut write = turn_request(false);
344                if let switchyard_protocol::ContentBlock::ToolCall(call) =
345                    &mut write.llm_request.messages[1].content[0]
346                {
347                    call.name = "Write".into();
348                }
349                for request in [
350                    initial,
351                    turn_request(true),
352                    write.clone(),
353                    write.clone(),
354                    write,
355                    turn_request(true),
356                    turn_request(true),
357                    turn_request(true),
358                ] {
359                    test_drive_with_models(
360                        router.clone(),
361                        request,
362                        runtime_models(),
363                        recorder.serve(),
364                    )
365                    .await?;
366                }
367                let actual: Vec<_> = recorder
368                    .routed()
369                    .into_iter()
370                    .map(|call| call.target)
371                    .collect();
372                let initial = if capable_first { "strong" } else { "weak" };
373                let mut expected = vec![initial, "strong", "strong", "strong", "weak"];
374                let count = turns.unwrap_or(if capable_first { -1 } else { 0 });
375                expected.extend((0..3).map(|i| {
376                    if count == -1 || i < count {
377                        "weak"
378                    } else {
379                        "strong"
380                    }
381                }));
382                assert_eq!(
383                    actual, expected,
384                    "capable_first={capable_first}, turns={turns:?}"
385                );
386                assert_eq!(recorder.judge_calls(), 1);
387            }
388        }
389        Ok(())
390    }
391
392    fn router_with_trigger(classify_trigger: ClassifyTrigger) -> Result<Arc<CompositeRouter>> {
393        router_with_stage(
394            classify_trigger,
395            StageRouterConfig::new(PickerMode::EfficientFirst, 0.5),
396        )
397    }
398
399    fn router_with_stage(
400        classify_trigger: ClassifyTrigger,
401        stage: StageRouterConfig,
402    ) -> Result<Arc<CompositeRouter>> {
403        Ok(Arc::new(CompositeRouter::new(CompositeRouterConfig {
404            judge: CompositeJudgeConfig::Capability(TaskClassifierConfig {
405                judge: CapabilityJudgeConfig::Llm(LlmCapabilityConfig {
406                    base_threshold: 0.5,
407                    ..LlmCapabilityConfig::default()
408                }),
409                classify_trigger,
410                ..Default::default()
411            }),
412            stage,
413        })?))
414    }
415
416    #[tokio::test]
417    async fn profiles_keep_their_own_holds_and_prompts_when_the_judge_switches() -> Result<()> {
418        for (has_session_id, picker, first_target, first_prompt) in [
419            (false, PickerMode::EfficientFirst, "weak", "EF efficient"),
420            (true, PickerMode::EfficientFirst, "weak", "EF efficient"),
421            (true, PickerMode::CapableFirst, "strong", "EF capable"),
422        ] {
423            let mut efficient = StageRouterConfig::new(picker, 0.5);
424            efficient.capable_hold_turns = -1;
425            efficient.capable_system_prompt = Some("EF capable".into());
426            efficient.efficient_system_prompt = Some("EF efficient".into());
427            let mut capable = StageRouterConfig::new(PickerMode::CapableFirst, 0.0);
428            capable.capable_hold_turns = 0;
429            capable.capable_system_prompt = Some("CF capable".into());
430            capable.efficient_system_prompt = Some("CF efficient".into());
431            let router = Arc::new(CompositeRouter::new_with_profiles(
432                CompositeJudgeConfig::Capability(TaskClassifierConfig {
433                    judge: CapabilityJudgeConfig::Llm(LlmCapabilityConfig {
434                        base_threshold: 0.5,
435                        ..Default::default()
436                    }),
437                    classify_trigger: ClassifyTrigger::UserTurn,
438                    ..Default::default()
439                }),
440                efficient,
441                capable,
442            )?);
443            let recorder = Arc::new(Recorder::default());
444            let mut write = turn_request(false);
445            if let switchyard_protocol::ContentBlock::ToolCall(call) =
446                &mut write.llm_request.messages[1].content[0]
447            {
448                call.name = "Write".into();
449            }
450            let mut state = State::default();
451            for (p_solve, request, target, prompt) in [
452                (1.0, user_turn_request(), first_target, first_prompt),
453                (1.0, turn_request(true), "strong", "EF capable"),
454                (0.0, user_turn_request(), "strong", "CF capable"),
455                (0.0, write, "weak", "CF efficient"),
456                (1.0, user_turn_request(), "strong", "EF capable"),
457                (0.0, user_turn_request(), "weak", "CF efficient"),
458            ] {
459                *recorder.judge_p_solve.lock() = p_solve;
460                let request = if has_session_id {
461                    request
462                } else {
463                    unkeyed(request)
464                };
465                let outcome = crate::core::testing::test_drive_with_state(
466                    router.clone(),
467                    request,
468                    runtime_models(),
469                    state,
470                    recorder.serve(),
471                )
472                .await?;
473                assert_eq!(outcome.selected_model_id()?.as_str(), target);
474                assert_eq!(outcome.request.llm_request.instructions.len(), 1);
475                assert_eq!(
476                    outcome.request.llm_request.instructions[0].content,
477                    vec![switchyard_protocol::ContentBlock::Text {
478                        text: prompt.into()
479                    }]
480                );
481                state = outcome.state;
482            }
483            assert_eq!(recorder.judge_calls(), 4);
484        }
485        Ok(())
486    }
487
488    #[test]
489    fn rejects_every_request_as_a_trigger() {
490        for judge in [
491            CompositeJudgeConfig::Capability(TaskClassifierConfig::default()),
492            CompositeJudgeConfig::Custom(CustomClassifierConfig::new(
493                "Choose a tier",
494                serde_json::json!({"type": "object"}),
495                crate::CustomClassifierPolicy::target_selector("/decision/target"),
496            )),
497        ] {
498            let error = CompositeRouter::new(CompositeRouterConfig {
499                judge,
500                stage: StageRouterConfig::new(PickerMode::EfficientFirst, 0.5),
501            })
502            .err()
503            .expect("every_request must be rejected for both classifier modes");
504            assert!(matches!(error, LibsyError::AlgorithmError { message }
505                if message == "composite: classify_trigger must be user_turn or new_session"));
506        }
507    }
508
509    fn custom_judge(trigger: ClassifyTrigger) -> CompositeJudgeConfig {
510        let mut config = CustomClassifierConfig::new(
511            "Select a tier",
512            serde_json::json!({"type": "object"}),
513            crate::CustomClassifierPolicy::target_selector("/decision/target"),
514        );
515        config.classify_trigger = trigger;
516        CompositeJudgeConfig::Custom(config)
517    }
518
519    #[tokio::test]
520    async fn a_failed_session_judgment_retains_the_configured_stage_default() -> Result<()> {
521        for custom in [false, true] {
522            for (mode, expected) in [
523                (PickerMode::EfficientFirst, "weak"),
524                (PickerMode::CapableFirst, "strong"),
525            ] {
526                let judge = if custom {
527                    custom_judge(ClassifyTrigger::NewSession)
528                } else {
529                    CompositeJudgeConfig::Capability(TaskClassifierConfig {
530                        classify_trigger: ClassifyTrigger::NewSession,
531                        ..Default::default()
532                    })
533                };
534                let router = Arc::new(CompositeRouter::new(CompositeRouterConfig {
535                    judge,
536                    stage: StageRouterConfig::new(mode, 0.5),
537                })?);
538                let calls = Arc::new(Mutex::new(0));
539                let serve = {
540                    let calls = calls.clone();
541                    move |target: ModelId, _request: Request| {
542                        let completion = if target.as_str() == JUDGE {
543                            *calls.lock() += 1;
544                            "not JSON".to_string()
545                        } else {
546                            target.to_string()
547                        };
548                        async move { Ok(crate::core::testing::reply(completion)) }
549                    }
550                };
551                for request in [
552                    user_turn_request(),
553                    turn_request(false),
554                    user_turn_request(),
555                ] {
556                    let (selected, _) = test_drive_with_models(
557                        router.clone(),
558                        request,
559                        runtime_models(),
560                        serve.clone(),
561                    )
562                    .await?;
563                    assert_eq!(selected.as_str(), expected);
564                }
565                assert_eq!(
566                    *calls.lock(),
567                    1,
568                    "new_session must retain a failed judgment"
569                );
570            }
571        }
572        Ok(())
573    }
574
575    #[tokio::test]
576    async fn a_custom_judge_sets_the_tier_and_retains_it_on_failure() -> Result<()> {
577        let router = Arc::new(CompositeRouter::new(CompositeRouterConfig {
578            judge: custom_judge(ClassifyTrigger::UserTurn),
579            stage: StageRouterConfig::new(PickerMode::EfficientFirst, 0.5),
580        })?);
581        let replies = Arc::new(Mutex::new(
582            [
583                Some(r#"{"decision":{"target":"capable"}}"#),
584                None,
585                Some("not JSON"),
586                Some(r#"{"decision":{"target":"efficient"}}"#),
587            ]
588            .into_iter(),
589        ));
590        let serve = {
591            let replies = replies.clone();
592            move |target: ModelId, _request: Request| {
593                let completion = if target.as_str() == JUDGE {
594                    replies
595                        .lock()
596                        .next()
597                        .expect("unexpected judge call")
598                        .map(str::to_string)
599                } else {
600                    Some(target.to_string())
601                };
602                async move {
603                    completion
604                        .map(crate::core::testing::reply)
605                        .ok_or_else(|| LlmClientError::General("judge unavailable".into()))
606                }
607            }
608        };
609        for (request, expected) in [
610            (user_turn_request(), "strong"),
611            (turn_request(false), "strong"),
612            (user_turn_request(), "strong"),
613            (user_turn_request(), "strong"),
614            (user_turn_request(), "weak"),
615        ] {
616            let (selected, _) =
617                test_drive_with_models(router.clone(), request, runtime_models(), serve.clone())
618                    .await?;
619            assert_eq!(selected.as_str(), expected);
620        }
621        assert!(replies.lock().next().is_none());
622        Ok(())
623    }
624
625    /// A route may point its judge at the same target it serves capable turns on.
626    /// The tier then cannot be recovered by looking the served model up in the
627    /// runtime groups — one id, two categories — so it comes off the verdict itself.
628    #[tokio::test]
629    async fn a_judge_sharing_the_capable_model_still_latches_the_tier() -> Result<()> {
630        let models: HashMap<Category, Vec<ModelId>> = [
631            (Category::Judge, vec![ModelId::from("strong")]),
632            (Category::Capable, vec![ModelId::from("strong")]),
633            (Category::Efficient, vec![ModelId::from("weak")]),
634            (
635                Category::Any,
636                vec![ModelId::from("strong"), ModelId::from("weak")],
637            ),
638        ]
639        .into();
640        // The judge runs first as a request-side processor, so the opening call is its own.
641        let calls = Arc::new(Mutex::new(0u32));
642        let serve = {
643            let calls = Arc::clone(&calls);
644            move |target: ModelId, _request: Request| {
645                let calls = Arc::clone(&calls);
646                async move {
647                    let mut calls = calls.lock();
648                    *calls += 1;
649                    let completion = if *calls == 1 {
650                        // p_solve below the threshold: the judge does not trust the
651                        // efficient tier, so the verdict is capable.
652                        r#"{"crux":"bounded task","primary_rule":"SUP-1","capability_boundary":"supported","p_solve":0.1}"#.to_string()
653                    } else {
654                        target.to_string()
655                    };
656                    Ok(crate::core::testing::reply(completion))
657                }
658            }
659        };
660        let router = router()?;
661
662        test_drive_with_models(
663            router.clone(),
664            user_turn_request(),
665            models.clone(),
666            serve.clone(),
667        )
668        .await?;
669        // A tool step is not a user turn, so nothing re-judges and the latched tier decides.
670        let (selected, _) =
671            test_drive_with_models(router, turn_request(false), models, serve).await?;
672
673        assert_eq!(selected, ModelId::from("strong"));
674        Ok(())
675    }
676
677    #[tokio::test]
678    async fn a_session_without_an_id_keys_on_the_message_hash() -> Result<()> {
679        let recorder = Arc::new(Recorder::default());
680        *recorder.judge_p_solve.lock() = 0.1;
681        let router = hash_keyed_router()?;
682
683        test_drive_with_models(
684            router.clone(),
685            unkeyed(user_turn_request()),
686            runtime_models(),
687            recorder.serve(),
688        )
689        .await?;
690        test_drive_with_models(
691            router.clone(),
692            unkeyed(turn_request(false)),
693            runtime_models(),
694            recorder.serve(),
695        )
696        .await?;
697
698        assert_eq!(
699            recorder.judge_calls(),
700            1,
701            "a tool step is not a user turn, session id or not"
702        );
703        assert_eq!(
704            recorder.routed()[1].target,
705            "strong",
706            "and the tier survives the tool step"
707        );
708        Ok(())
709    }
710
711    #[tokio::test]
712    async fn the_judge_sets_the_tier_once_a_turn_and_the_signals_run_within_it() -> Result<()> {
713        let recorder = Arc::new(Recorder::default());
714        *recorder.judge_p_solve.lock() = 0.1;
715        let router = router()?;
716
717        test_drive_with_models(
718            router.clone(),
719            user_turn_request(),
720            runtime_models(),
721            recorder.serve(),
722        )
723        .await?;
724        test_drive_with_models(
725            router.clone(),
726            turn_request(false),
727            runtime_models(),
728            recorder.serve(),
729        )
730        .await?;
731
732        let routed = recorder.routed();
733        assert_eq!(
734            routed[0].target, "strong",
735            "a quiet turn falls open to the verdict"
736        );
737        assert_eq!(
738            routed[1].target, "strong",
739            "which holds across the tool steps after it"
740        );
741        assert_eq!(
742            recorder.judge_calls(),
743            1,
744            "a tool step is not a new user turn"
745        );
746        Ok(())
747    }
748
749    #[tokio::test]
750    async fn caller_state_carries_the_judged_tier_without_a_session_id() -> Result<()> {
751        for trigger in [ClassifyTrigger::UserTurn, ClassifyTrigger::NewSession] {
752            let recorder = Arc::new(Recorder::default());
753            *recorder.judge_p_solve.lock() = 0.1;
754            let router = router_with_trigger(trigger)?;
755            let run = |request, state| {
756                crate::core::testing::test_drive_with_state(
757                    router.clone(),
758                    request,
759                    runtime_models(),
760                    state,
761                    recorder.serve(),
762                )
763            };
764
765            let first = run(unkeyed(user_turn_request()), State::default()).await?;
766            assert_eq!(first.selected_model_id()?, &ModelId::from("strong"));
767
768            let continued = run(unkeyed(turn_request(false)), first.state).await?;
769            assert_eq!(continued.selected_model_id()?, &ModelId::from("strong"));
770            assert_eq!(
771                continued.metadata.and_then(|m| m.evidence),
772                Some(serde_json::json!({"source": "retained"}))
773            );
774            assert_eq!(recorder.judge_calls(), 1, "{trigger:?}");
775        }
776        Ok(())
777    }
778}
switchyard_libsy/algorithms/composite.rs

switchyard_libsy/algorithms/
composite.rs
