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

switchyard_libsy/core/
processor.rs
