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

switchyard_libsy/core/
processor.rs
