orchestral_runtime/
concurrency.rs

1//! Concurrency Policy
2//!
3//! Defines how the runtime handles new inputs when an interaction is already running.
4
5use orchestral_core::store::Event;
6
7/// Decision on how to handle concurrent interactions
8#[derive(Debug, Clone, PartialEq, Eq)]
9pub enum ConcurrencyDecision {
10    /// Interrupt the current interaction and start a new one
11    InterruptAndStartNew,
12    /// Reject the new input
13    Reject {
14        /// Reason for rejection
15        reason: String,
16    },
17    /// Queue the new input for later processing
18    Queue,
19    /// Run in parallel with the current interaction
20    Parallel,
21    /// Merge the new input into the running interaction
22    MergeIntoRunning,
23}
24
25impl ConcurrencyDecision {
26    /// Create a reject decision
27    pub fn reject(reason: impl Into<String>) -> Self {
28        Self::Reject {
29            reason: reason.into(),
30        }
31    }
32}
33
34/// Running state information for concurrency decisions
35#[derive(Debug, Clone)]
36pub struct RunningState {
37    /// Number of currently active interactions
38    pub active_count: usize,
39    /// Whether any interaction is currently processing
40    pub is_processing: bool,
41    /// Whether any interaction is waiting for user input
42    pub is_waiting_user: bool,
43    /// Whether any interaction is waiting for external event
44    pub is_waiting_event: bool,
45}
46
47impl RunningState {
48    /// Create a new running state
49    pub fn new() -> Self {
50        Self {
51            active_count: 0,
52            is_processing: false,
53            is_waiting_user: false,
54            is_waiting_event: false,
55        }
56    }
57
58    /// Check if there are any active interactions
59    pub fn has_active(&self) -> bool {
60        self.active_count > 0
61    }
62
63    /// Check if the runtime is idle
64    pub fn is_idle(&self) -> bool {
65        self.active_count == 0
66    }
67}
68
69impl Default for RunningState {
70    fn default() -> Self {
71        Self::new()
72    }
73}
74
75/// Concurrency policy trait
76///
77/// Implementations decide how to handle new inputs based on the current running state.
78pub trait ConcurrencyPolicy: Send + Sync {
79    /// Decide how to handle a new event given the current running state
80    fn decide(&self, running: &RunningState, new_event: &Event) -> ConcurrencyDecision;
81}
82
83/// Default concurrency policy: Interrupt and start new
84///
85/// This is the most common behavior for chat-like applications.
86pub struct DefaultConcurrencyPolicy;
87
88impl ConcurrencyPolicy for DefaultConcurrencyPolicy {
89    fn decide(&self, running: &RunningState, _new_event: &Event) -> ConcurrencyDecision {
90        if running.is_idle() {
91            // No active interaction, start new
92            ConcurrencyDecision::InterruptAndStartNew
93        } else {
94            // Interrupt current and start new
95            ConcurrencyDecision::InterruptAndStartNew
96        }
97    }
98}
99
100/// Queue policy placeholder.
101///
102/// Runtime-side queue execution is not implemented yet, so this policy currently
103/// rejects when busy with an explicit reason.
104pub struct QueueConcurrencyPolicy;
105
106impl ConcurrencyPolicy for QueueConcurrencyPolicy {
107    fn decide(&self, running: &RunningState, _new_event: &Event) -> ConcurrencyDecision {
108        if running.is_idle() {
109            ConcurrencyDecision::InterruptAndStartNew
110        } else {
111            ConcurrencyDecision::reject(
112                "Queue policy is not implemented; use interrupt/parallel/reject policy",
113            )
114        }
115    }
116}
117
118/// Parallel policy: Allow multiple interactions to run in parallel
119pub struct ParallelConcurrencyPolicy {
120    /// Maximum number of parallel interactions
121    pub max_parallel: usize,
122}
123
124impl ParallelConcurrencyPolicy {
125    /// Create a new parallel policy
126    pub fn new(max_parallel: usize) -> Self {
127        Self { max_parallel }
128    }
129}
130
131impl Default for ParallelConcurrencyPolicy {
132    fn default() -> Self {
133        Self::new(4)
134    }
135}
136
137impl ConcurrencyPolicy for ParallelConcurrencyPolicy {
138    fn decide(&self, running: &RunningState, _new_event: &Event) -> ConcurrencyDecision {
139        if running.active_count >= self.max_parallel {
140            ConcurrencyDecision::reject(format!(
141                "Maximum parallel interactions ({}) reached",
142                self.max_parallel
143            ))
144        } else {
145            ConcurrencyDecision::Parallel
146        }
147    }
148}
149
150/// Merge policy: Merge new inputs into the running interaction.
151///
152/// Best for chat bots — new messages join the current conversation turn.
153/// If no interaction is active, starts a new one.
154pub struct MergeConcurrencyPolicy;
155
156impl ConcurrencyPolicy for MergeConcurrencyPolicy {
157    fn decide(&self, running: &RunningState, _new_event: &Event) -> ConcurrencyDecision {
158        if running.is_idle() {
159            ConcurrencyDecision::InterruptAndStartNew
160        } else {
161            ConcurrencyDecision::MergeIntoRunning
162        }
163    }
164}
165
166/// Reject policy: Reject new inputs when busy
167pub struct RejectWhenBusyConcurrencyPolicy;
168
169impl ConcurrencyPolicy for RejectWhenBusyConcurrencyPolicy {
170    fn decide(&self, running: &RunningState, _new_event: &Event) -> ConcurrencyDecision {
171        if running.is_idle() {
172            ConcurrencyDecision::InterruptAndStartNew
173        } else {
174            ConcurrencyDecision::reject("System is busy processing another request")
175        }
176    }
177}
orchestral_runtime/concurrency.rs

orchestral_runtime/
concurrency.rs
