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        category: Option<Category>,
34        /// Offered so a processor can inspect the runtime model categories.
35        driver: &'a Driver,
36    },
37    /// A buffered response received back from a model.
38    ModelResponse(&'a AggLlmResponse),
39}
40
41/// Collects events as the algorithm runs and mutates the composition's state.
42#[async_trait]
43pub trait Processor<S = ()>: Send + Sync {
44    /// Process an event, accumulating facts into `state`. Request-bearing events
45    /// ([`Event::Request`], [`Event::Decision`]) may also be rewritten in place.
46    async fn process(&self, state: &mut S, event: Event<'_>) -> Result<()>;
47}
48
49#[cfg(test)]
50mod tests {
51    use super::*;
52    use crate::core::testing::empty_driver;
53    use std::collections::HashMap;
54    use switchyard_protocol::{text_request, text_response};
55
56    type TestState = HashMap<&'static str, u32>;
57
58    /// The key each event variant tallies under.
59    fn event_key(event: &Event<'_>) -> &'static str {
60        match event {
61            Event::Request { .. } => "requests",
62            Event::Decision { .. } => "decisions",
63            Event::ModelResponse(_) => "model_responses",
64        }
65    }
66
67    /// Reads a count, treating a missing key as zero.
68    fn count(state: &TestState, key: &'static str) -> u32 {
69        state.get(key).copied().unwrap_or_default()
70    }
71
72    /// Tallies each event variant under its own key.
73    struct CountingProcessor;
74
75    #[async_trait]
76    impl Processor<TestState> for CountingProcessor {
77        async fn process(&self, state: &mut TestState, event: Event<'_>) -> Result<()> {
78            *state.entry(event_key(&event)).or_default() += 1;
79            Ok(())
80        }
81    }
82
83    fn request() -> Request {
84        Request {
85            llm_request: text_request(Some("auto".to_string()), "hi"),
86            raw_request: None,
87            metadata: None,
88        }
89    }
90
91    #[tokio::test]
92    async fn processor_tallies_each_event_variant_into_state() -> Result<()> {
93        let processor = CountingProcessor;
94        let mut state = TestState::default();
95        let mut req = request();
96        let response = text_response(None, "ok");
97        let selected_model_id = ModelId::from("test/model");
98        // Feed one of every event variant through the processor.
99        processor
100            .process(
101                &mut state,
102                Event::Request {
103                    request: &mut req,
104                    driver: &empty_driver(),
105                },
106            )
107            .await?;
108        processor
109            .process(&mut state, Event::ModelResponse(&response))
110            .await?;
111        processor
112            .process(
113                &mut state,
114                Event::Decision {
115                    request: &mut req,
116                    selected_model_id: &selected_model_id,
117                    category: None,
118                    driver: &empty_driver(),
119                },
120            )
121            .await?;
122        assert_eq!(count(&state, "requests"), 1);
123        assert_eq!(count(&state, "decisions"), 1);
124        assert_eq!(count(&state, "model_responses"), 1);
125        Ok(())
126    }
127
128    #[tokio::test]
129    async fn process_accumulates_state_across_repeated_events() -> Result<()> {
130        let processor = CountingProcessor;
131        let mut state = TestState::default();
132        let mut req = request();
133
134        for _ in 0..3 {
135            processor
136                .process(
137                    &mut state,
138                    Event::Request {
139                        request: &mut req,
140                        driver: &empty_driver(),
141                    },
142                )
143                .await?;
144        }
145
146        assert_eq!(count(&state, "requests"), 3);
147        Ok(())
148    }
149
150    /// Rewrites the requested model on every request-bearing event.
151    struct RewritingProcessor;
152
153    #[async_trait]
154    impl Processor for RewritingProcessor {
155        async fn process(&self, _state: &mut (), event: Event<'_>) -> Result<()> {
156            match event {
157                Event::Request { request, .. } | Event::Decision { request, .. } => {
158                    request.llm_request.model = Some("rewritten".to_string());
159                }
160                _ => {}
161            }
162            Ok(())
163        }
164    }
165
166    #[tokio::test]
167    async fn processor_rewrites_the_request_in_place() -> Result<()> {
168        let mut state = ();
169        let mut req = request();
170        assert_eq!(req.model_id(), Some("auto".into()));
171
172        RewritingProcessor
173            .process(
174                &mut state,
175                Event::Request {
176                    request: &mut req,
177                    driver: &empty_driver(),
178                },
179            )
180            .await?;
181
182        // The edit outlives the call, so the next component sees the rewritten request.
183        assert_eq!(req.model_id(), Some("rewritten".into()));
184        Ok(())
185    }
186}
switchyard_libsy/core/processor.rs

switchyard_libsy/core/
processor.rs
