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

meerkat_runtime/
coalescing.rs
