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

switchyard_libsy/core/
processor.rs
