meerkat_runtime/
coalescing.rs

1//! §16 Coalescing — merging compatible inputs into aggregates.
2//!
3//! Only ExternalEventInput and PeerInput(ResponseProgress) are coalescing-eligible.
4//! Supersession scoped by (LogicalRuntimeId, variant, SupersessionKey).
5//! Cross-kind forbidden.
6
7use chrono::Utc;
8use meerkat_core::lifecycle::InputId;
9
10use crate::identifiers::{InputKind, LogicalRuntimeId, SupersessionKey};
11use crate::input::{Input, PeerConvention};
12
13/// Whether an input is eligible for coalescing.
14pub fn is_coalescing_eligible(input: &Input) -> bool {
15    matches!(
16        input,
17        Input::ExternalEvent(_)
18            | Input::Peer(crate::input::PeerInput {
19                convention: Some(PeerConvention::ResponseProgress { .. }),
20                ..
21            })
22    )
23}
24
25/// Scope key for supersession — inputs with the same scope can supersede each other.
26#[derive(Debug, Clone, PartialEq, Eq, Hash)]
27pub struct SupersessionScope {
28    pub runtime_id: LogicalRuntimeId,
29    pub kind: InputKind,
30    pub supersession_key: SupersessionKey,
31}
32
33impl SupersessionScope {
34    /// Extract the supersession scope from an input (if it has a supersession key).
35    pub fn from_input(input: &Input, runtime_id: &LogicalRuntimeId) -> Option<Self> {
36        let key = input.header().supersession_key.as_ref()?;
37        Some(Self {
38            runtime_id: runtime_id.clone(),
39            kind: input.kind(),
40            supersession_key: key.clone(),
41        })
42    }
43}
44
45/// Result of a coalescing check.
46#[derive(Debug)]
47pub enum CoalescingResult {
48    /// No coalescing needed (input is standalone).
49    Standalone,
50    /// The input supersedes an existing input.
51    Supersedes {
52        /// ID of the superseded input.
53        superseded_id: InputId,
54    },
55}
56
57/// Check if a new input supersedes an existing input with the same scope.
58pub fn check_supersession(
59    new_input: &Input,
60    existing_input: &Input,
61    runtime_id: &LogicalRuntimeId,
62) -> CoalescingResult {
63    let new_scope = SupersessionScope::from_input(new_input, runtime_id);
64    let existing_scope = SupersessionScope::from_input(existing_input, runtime_id);
65
66    match (new_scope, existing_scope) {
67        (Some(ns), Some(es)) if ns == es => {
68            // Same scope — new supersedes existing
69            CoalescingResult::Supersedes {
70                superseded_id: existing_input.id().clone(),
71            }
72        }
73        _ => CoalescingResult::Standalone,
74    }
75}
76
77/// Create an aggregate input from multiple coalesced inputs.
78pub fn create_aggregate_input(
79    sources: &[&Input],
80    aggregate_id: InputId,
81) -> Option<AggregateDescriptor> {
82    if sources.is_empty() {
83        return None;
84    }
85
86    let source_ids: Vec<InputId> = sources.iter().map(|i| i.id().clone()).collect();
87    let summary = format!("{} coalesced inputs", sources.len());
88
89    Some(AggregateDescriptor {
90        aggregate_id,
91        source_ids,
92        summary,
93        created_at: Utc::now(),
94    })
95}
96
97/// Describes a coalesced aggregate (the caller creates the actual Input).
98#[derive(Debug, Clone)]
99pub struct AggregateDescriptor {
100    pub aggregate_id: InputId,
101    pub source_ids: Vec<InputId>,
102    pub summary: String,
103    pub created_at: chrono::DateTime<Utc>,
104}
105
106#[cfg(test)]
107#[allow(clippy::unwrap_used)]
108mod tests {
109    use super::*;
110    use crate::input::*;
111    use chrono::Utc;
112    use meerkat_core::types::HandlingMode;
113
114    fn make_header_with_supersession(key: Option<&str>) -> InputHeader {
115        InputHeader {
116            id: InputId::new(),
117            timestamp: Utc::now(),
118            source: InputOrigin::External {
119                source_name: "test".into(),
120            },
121            durability: InputDurability::Ephemeral,
122            visibility: InputVisibility::default(),
123            idempotency_key: None,
124            supersession_key: key.map(SupersessionKey::new),
125            correlation_id: None,
126        }
127    }
128
129    #[test]
130    fn external_event_is_coalescing_eligible() {
131        let input = Input::ExternalEvent(ExternalEventInput {
132            objective_id: None,
133            header: make_header_with_supersession(None),
134            event_type: "webhook".into(),
135            payload: serde_json::json!({}),
136            blocks: None,
137            handling_mode: HandlingMode::Queue,
138            render_metadata: None,
139        });
140        assert!(is_coalescing_eligible(&input));
141    }
142
143    #[test]
144    fn response_progress_is_coalescing_eligible() {
145        let input = Input::Peer(PeerInput {
146            directed_interaction_id: None,
147            objective_id: None,
148            injected_context: Vec::new(),
149            sender_taint: None,
150            header: make_header_with_supersession(None),
151            convention: Some(PeerConvention::ResponseProgress {
152                request_id: "req-1".into(),
153                phase: ResponseProgressPhase::InProgress,
154            }),
155            content: "progress".into(),
156            payload: Some(serde_json::json!({"progress": "working"})),
157            handling_mode: None,
158        });
159        assert!(is_coalescing_eligible(&input));
160    }
161
162    #[test]
163    fn prompt_not_coalescing_eligible() {
164        let input = Input::Prompt(PromptInput {
165            injected_context: Vec::new(),
166            header: make_header_with_supersession(None),
167            content: "hello".into(),
168            typed_turn_appends: Vec::new(),
169            turn_metadata: None,
170        });
171        assert!(!is_coalescing_eligible(&input));
172    }
173
174    #[test]
175    fn peer_message_not_coalescing_eligible() {
176        let input = Input::Peer(PeerInput {
177            directed_interaction_id: None,
178            objective_id: None,
179            injected_context: Vec::new(),
180            sender_taint: None,
181            header: make_header_with_supersession(None),
182            convention: Some(PeerConvention::Message),
183            content: "hello".into(),
184            payload: None,
185            handling_mode: None,
186        });
187        assert!(!is_coalescing_eligible(&input));
188    }
189
190    #[test]
191    fn supersession_same_scope() {
192        let runtime = LogicalRuntimeId::new("agent-1");
193        let input1 = Input::ExternalEvent(ExternalEventInput {
194            objective_id: None,
195            header: make_header_with_supersession(Some("status")),
196            event_type: "status_update".into(),
197            payload: serde_json::json!({"v": 1}),
198            blocks: None,
199            handling_mode: HandlingMode::Queue,
200            render_metadata: None,
201        });
202        let input2 = Input::ExternalEvent(ExternalEventInput {
203            objective_id: None,
204            header: make_header_with_supersession(Some("status")),
205            event_type: "status_update".into(),
206            payload: serde_json::json!({"v": 2}),
207            blocks: None,
208            handling_mode: HandlingMode::Queue,
209            render_metadata: None,
210        });
211        let result = check_supersession(&input2, &input1, &runtime);
212        assert!(matches!(result, CoalescingResult::Supersedes { .. }));
213    }
214
215    #[test]
216    fn supersession_different_key() {
217        let runtime = LogicalRuntimeId::new("agent-1");
218        let input1 = Input::ExternalEvent(ExternalEventInput {
219            objective_id: None,
220            header: make_header_with_supersession(Some("status-a")),
221            event_type: "status_update".into(),
222            payload: serde_json::json!({}),
223            blocks: None,
224            handling_mode: HandlingMode::Queue,
225            render_metadata: None,
226        });
227        let input2 = Input::ExternalEvent(ExternalEventInput {
228            objective_id: None,
229            header: make_header_with_supersession(Some("status-b")),
230            event_type: "status_update".into(),
231            payload: serde_json::json!({}),
232            blocks: None,
233            handling_mode: HandlingMode::Queue,
234            render_metadata: None,
235        });
236        let result = check_supersession(&input2, &input1, &runtime);
237        assert!(matches!(result, CoalescingResult::Standalone));
238    }
239
240    #[test]
241    fn supersession_no_key() {
242        let runtime = LogicalRuntimeId::new("agent-1");
243        let input1 = Input::ExternalEvent(ExternalEventInput {
244            objective_id: None,
245            header: make_header_with_supersession(None),
246            event_type: "status_update".into(),
247            payload: serde_json::json!({}),
248            blocks: None,
249            handling_mode: HandlingMode::Queue,
250            render_metadata: None,
251        });
252        let input2 = Input::ExternalEvent(ExternalEventInput {
253            objective_id: None,
254            header: make_header_with_supersession(None),
255            event_type: "status_update".into(),
256            payload: serde_json::json!({}),
257            blocks: None,
258            handling_mode: HandlingMode::Queue,
259            render_metadata: None,
260        });
261        let result = check_supersession(&input2, &input1, &runtime);
262        assert!(matches!(result, CoalescingResult::Standalone));
263    }
264
265    #[test]
266    fn cross_kind_supersession_forbidden() {
267        let runtime = LogicalRuntimeId::new("agent-1");
268        let input1 = Input::ExternalEvent(ExternalEventInput {
269            objective_id: None,
270            header: make_header_with_supersession(Some("same-key")),
271            event_type: "type-a".into(),
272            payload: serde_json::json!({}),
273            blocks: None,
274            handling_mode: HandlingMode::Queue,
275            render_metadata: None,
276        });
277        // Different kind (Prompt vs ExternalEvent) but same supersession key
278        let input2 = Input::Prompt(PromptInput {
279            injected_context: Vec::new(),
280            header: make_header_with_supersession(Some("same-key")),
281            content: "hello".into(),
282            typed_turn_appends: Vec::new(),
283            turn_metadata: None,
284        });
285        let result = check_supersession(&input2, &input1, &runtime);
286        // Different kinds → different scope → no supersession
287        assert!(matches!(result, CoalescingResult::Standalone));
288    }
289
290    #[test]
291    fn create_aggregate_from_sources() {
292        let sources: Vec<Input> = (0..3)
293            .map(|_| {
294                Input::ExternalEvent(ExternalEventInput {
295                    objective_id: None,
296                    header: make_header_with_supersession(None),
297                    event_type: "test".into(),
298                    payload: serde_json::json!({}),
299                    blocks: None,
300                    handling_mode: HandlingMode::Queue,
301                    render_metadata: None,
302                })
303            })
304            .collect();
305        let source_refs: Vec<&Input> = sources.iter().collect();
306        let agg = create_aggregate_input(&source_refs, InputId::new()).unwrap();
307        assert_eq!(agg.source_ids.len(), 3);
308        assert!(agg.summary.contains('3'));
309    }
310
311    #[test]
312    fn create_aggregate_empty_returns_none() {
313        let result = create_aggregate_input(&[], InputId::new());
314        assert!(result.is_none());
315    }
316}
meerkat_runtime/coalescing.rs

meerkat_runtime/
coalescing.rs
