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

switchyard_libsy/core/
processor.rs
