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            objective_id: None,
147            injected_context: Vec::new(),
148            sender_taint: None,
149            header: make_header_with_supersession(None),
150            convention: Some(PeerConvention::ResponseProgress {
151                request_id: "req-1".into(),
152                phase: ResponseProgressPhase::InProgress,
153            }),
154            content: "progress".into(),
155            payload: Some(serde_json::json!({"progress": "working"})),
156            handling_mode: None,
157        });
158        assert!(is_coalescing_eligible(&input));
159    }
160
161    #[test]
162    fn prompt_not_coalescing_eligible() {
163        let input = Input::Prompt(PromptInput {
164            injected_context: Vec::new(),
165            header: make_header_with_supersession(None),
166            content: "hello".into(),
167            typed_turn_appends: Vec::new(),
168            turn_metadata: None,
169        });
170        assert!(!is_coalescing_eligible(&input));
171    }
172
173    #[test]
174    fn peer_message_not_coalescing_eligible() {
175        let input = Input::Peer(PeerInput {
176            objective_id: None,
177            injected_context: Vec::new(),
178            sender_taint: None,
179            header: make_header_with_supersession(None),
180            convention: Some(PeerConvention::Message),
181            content: "hello".into(),
182            payload: None,
183            handling_mode: None,
184        });
185        assert!(!is_coalescing_eligible(&input));
186    }
187
188    #[test]
189    fn supersession_same_scope() {
190        let runtime = LogicalRuntimeId::new("agent-1");
191        let input1 = Input::ExternalEvent(ExternalEventInput {
192            objective_id: None,
193            header: make_header_with_supersession(Some("status")),
194            event_type: "status_update".into(),
195            payload: serde_json::json!({"v": 1}),
196            blocks: None,
197            handling_mode: HandlingMode::Queue,
198            render_metadata: None,
199        });
200        let input2 = Input::ExternalEvent(ExternalEventInput {
201            objective_id: None,
202            header: make_header_with_supersession(Some("status")),
203            event_type: "status_update".into(),
204            payload: serde_json::json!({"v": 2}),
205            blocks: None,
206            handling_mode: HandlingMode::Queue,
207            render_metadata: None,
208        });
209        let result = check_supersession(&input2, &input1, &runtime);
210        assert!(matches!(result, CoalescingResult::Supersedes { .. }));
211    }
212
213    #[test]
214    fn supersession_different_key() {
215        let runtime = LogicalRuntimeId::new("agent-1");
216        let input1 = Input::ExternalEvent(ExternalEventInput {
217            objective_id: None,
218            header: make_header_with_supersession(Some("status-a")),
219            event_type: "status_update".into(),
220            payload: serde_json::json!({}),
221            blocks: None,
222            handling_mode: HandlingMode::Queue,
223            render_metadata: None,
224        });
225        let input2 = Input::ExternalEvent(ExternalEventInput {
226            objective_id: None,
227            header: make_header_with_supersession(Some("status-b")),
228            event_type: "status_update".into(),
229            payload: serde_json::json!({}),
230            blocks: None,
231            handling_mode: HandlingMode::Queue,
232            render_metadata: None,
233        });
234        let result = check_supersession(&input2, &input1, &runtime);
235        assert!(matches!(result, CoalescingResult::Standalone));
236    }
237
238    #[test]
239    fn supersession_no_key() {
240        let runtime = LogicalRuntimeId::new("agent-1");
241        let input1 = Input::ExternalEvent(ExternalEventInput {
242            objective_id: None,
243            header: make_header_with_supersession(None),
244            event_type: "status_update".into(),
245            payload: serde_json::json!({}),
246            blocks: None,
247            handling_mode: HandlingMode::Queue,
248            render_metadata: None,
249        });
250        let input2 = Input::ExternalEvent(ExternalEventInput {
251            objective_id: None,
252            header: make_header_with_supersession(None),
253            event_type: "status_update".into(),
254            payload: serde_json::json!({}),
255            blocks: None,
256            handling_mode: HandlingMode::Queue,
257            render_metadata: None,
258        });
259        let result = check_supersession(&input2, &input1, &runtime);
260        assert!(matches!(result, CoalescingResult::Standalone));
261    }
262
263    #[test]
264    fn cross_kind_supersession_forbidden() {
265        let runtime = LogicalRuntimeId::new("agent-1");
266        let input1 = Input::ExternalEvent(ExternalEventInput {
267            objective_id: None,
268            header: make_header_with_supersession(Some("same-key")),
269            event_type: "type-a".into(),
270            payload: serde_json::json!({}),
271            blocks: None,
272            handling_mode: HandlingMode::Queue,
273            render_metadata: None,
274        });
275        // Different kind (Prompt vs ExternalEvent) but same supersession key
276        let input2 = Input::Prompt(PromptInput {
277            injected_context: Vec::new(),
278            header: make_header_with_supersession(Some("same-key")),
279            content: "hello".into(),
280            typed_turn_appends: Vec::new(),
281            turn_metadata: None,
282        });
283        let result = check_supersession(&input2, &input1, &runtime);
284        // Different kinds → different scope → no supersession
285        assert!(matches!(result, CoalescingResult::Standalone));
286    }
287
288    #[test]
289    fn create_aggregate_from_sources() {
290        let sources: Vec<Input> = (0..3)
291            .map(|_| {
292                Input::ExternalEvent(ExternalEventInput {
293                    objective_id: None,
294                    header: make_header_with_supersession(None),
295                    event_type: "test".into(),
296                    payload: serde_json::json!({}),
297                    blocks: None,
298                    handling_mode: HandlingMode::Queue,
299                    render_metadata: None,
300                })
301            })
302            .collect();
303        let source_refs: Vec<&Input> = sources.iter().collect();
304        let agg = create_aggregate_input(&source_refs, InputId::new()).unwrap();
305        assert_eq!(agg.source_ids.len(), 3);
306        assert!(agg.summary.contains('3'));
307    }
308
309    #[test]
310    fn create_aggregate_empty_returns_none() {
311        let result = create_aggregate_input(&[], InputId::new());
312        assert!(result.is_none());
313    }
314}
meerkat_runtime/coalescing.rs

meerkat_runtime/
coalescing.rs
