switchyard_libsy/core/
processor.rs

1// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
2// SPDX-License-Identifier: Apache-2.0
3
4use crate::Result;
5use crate::core::algorithm::Driver;
6use async_trait::async_trait;
7use switchyard_protocol::{AggLlmResponse, Category, ModelId, Request};
8
9/// An event observed by the algorithm. Events are consumed by [`Processor`] to mutate state.
10///
11/// Request-bearing variants ([`Event::Request`], [`Event::Decision`]) borrow the request
12/// mutably, so a processor may rewrite it in place and the edit propagates to the rest of
13/// the chain and to the model call. The observation-only variants stay immutable.
14pub enum Event<'a> {
15    /// The inbound request that begins a turn.
16    Request {
17        /// The request, rewritable in place.
18        request: &'a mut Request,
19        /// Offered so a processor can consult a model before the cascade runs.
20        driver: &'a Driver,
21    },
22    /// A routing decision paired with the request that produced it.
23    ///
24    /// The request is rewritable: a processor may add instructions or notes here that
25    /// are bound to the routing outcome (e.g. a model-specific system prompt).
26    Decision {
27        /// The request, rewritable in place.
28        request: &'a mut Request,
29        /// The model selected for `request`.
30        selected_model_id: &'a ModelId,
31        /// The category `selected_model_id` was drawn from, when the deciding
32        /// classifier picked one. `None` for a decision made without a category,
33        /// such as an affinity replay.
34        category: Option<Category>,
35        /// Offered so a processor can inspect the runtime model categories.
36        driver: &'a Driver,
37    },
38    /// A buffered response received back from a model.
39    ModelResponse(&'a AggLlmResponse),
40}
41
42/// Collects events as the algorithm runs and mutates the composition's state.
43#[async_trait]
44pub trait Processor<S = ()>: Send + Sync {
45    /// Process an event, accumulating facts into `state`. Request-bearing events
46    /// ([`Event::Request`], [`Event::Decision`]) may also be rewritten in place.
47    async fn process(&self, state: &mut S, event: Event<'_>) -> Result<()>;
48}
49
50#[cfg(test)]
51mod tests {
52    use super::*;
53    use crate::core::testing::empty_driver;
54    use std::collections::HashMap;
55    use switchyard_protocol::{text_request, text_response};
56
57    type TestState = HashMap<&'static str, u32>;
58
59    /// The key each event variant tallies under.
60    fn event_key(event: &Event<'_>) -> &'static str {
61        match event {
62            Event::Request { .. } => "requests",
63            Event::Decision { .. } => "decisions",
64            Event::ModelResponse(_) => "model_responses",
65        }
66    }
67
68    /// Reads a count, treating a missing key as zero.
69    fn count(state: &TestState, key: &'static str) -> u32 {
70        state.get(key).copied().unwrap_or_default()
71    }
72
73    /// Tallies each event variant under its own key.
74    struct CountingProcessor;
75
76    #[async_trait]
77    impl Processor<TestState> for CountingProcessor {
78        async fn process(&self, state: &mut TestState, event: Event<'_>) -> Result<()> {
79            *state.entry(event_key(&event)).or_default() += 1;
80            Ok(())
81        }
82    }
83
84    fn request() -> Request {
85        Request {
86            llm_request: text_request(Some("auto".to_string()), "hi"),
87            raw_request: None,
88            metadata: None,
89        }
90    }
91
92    #[tokio::test]
93    async fn processor_tallies_each_event_variant_into_state() -> Result<()> {
94        let processor = CountingProcessor;
95        let mut state = TestState::default();
96        let mut req = request();
97        let response = text_response(None, "ok");
98        let selected_model_id = ModelId::from("test/model");
99        // Feed one of every event variant through the processor.
100        processor
101            .process(
102                &mut state,
103                Event::Request {
104                    request: &mut req,
105                    driver: &empty_driver(),
106                },
107            )
108            .await?;
109        processor
110            .process(&mut state, Event::ModelResponse(&response))
111            .await?;
112        processor
113            .process(
114                &mut state,
115                Event::Decision {
116                    request: &mut req,
117                    selected_model_id: &selected_model_id,
118                    category: None,
119                    driver: &empty_driver(),
120                },
121            )
122            .await?;
123        assert_eq!(count(&state, "requests"), 1);
124        assert_eq!(count(&state, "decisions"), 1);
125        assert_eq!(count(&state, "model_responses"), 1);
126        Ok(())
127    }
128
129    #[tokio::test]
130    async fn process_accumulates_state_across_repeated_events() -> Result<()> {
131        let processor = CountingProcessor;
132        let mut state = TestState::default();
133        let mut req = request();
134
135        for _ in 0..3 {
136            processor
137                .process(
138                    &mut state,
139                    Event::Request {
140                        request: &mut req,
141                        driver: &empty_driver(),
142                    },
143                )
144                .await?;
145        }
146
147        assert_eq!(count(&state, "requests"), 3);
148        Ok(())
149    }
150
151    /// Rewrites the requested model on every request-bearing event.
152    struct RewritingProcessor;
153
154    #[async_trait]
155    impl Processor for RewritingProcessor {
156        async fn process(&self, _state: &mut (), event: Event<'_>) -> Result<()> {
157            match event {
158                Event::Request { request, .. } | Event::Decision { request, .. } => {
159                    request.llm_request.model = Some("rewritten".to_string());
160                }
161                _ => {}
162            }
163            Ok(())
164        }
165    }
166
167    #[tokio::test]
168    async fn processor_rewrites_the_request_in_place() -> Result<()> {
169        let mut state = ();
170        let mut req = request();
171        assert_eq!(req.model_id(), Some("auto".into()));
172
173        RewritingProcessor
174            .process(
175                &mut state,
176                Event::Request {
177                    request: &mut req,
178                    driver: &empty_driver(),
179                },
180            )
181            .await?;
182
183        // The edit outlives the call, so the next component sees the rewritten request.
184        assert_eq!(req.model_id(), Some("rewritten".into()));
185        Ok(())
186    }
187}
switchyard_libsy/core/processor.rs

switchyard_libsy/core/
processor.rs
