meerkat_mobkit/runtime/
event_transport.rs

1//! Event normalization and transport — line parsing, source validation, and envelope construction.
2
3use super::*;
4
5pub fn normalize_event_line(line: &str) -> Result<EventEnvelope<UnifiedEvent>, NormalizationError> {
6    if let Ok(envelope) = parse_unified_event_line(line) {
7        return enforce_source_consistency(envelope);
8    }
9
10    let value: Value = serde_json::from_str(line).map_err(|_| NormalizationError::InvalidJson)?;
11    let object = value.as_object().ok_or(NormalizationError::InvalidSchema)?;
12
13    let event_id = required_string(object.get("event_id"), "event_id")?;
14    let source = required_string(object.get("source"), "source")?;
15    let timestamp_ms = required_u64(object.get("timestamp_ms"), "timestamp_ms")?;
16
17    if let Some(module) = object.get("module") {
18        let module = required_string(Some(module), "module")?;
19        let event_type = required_string(object.get("event_type"), "event_type")?;
20        let payload = object
21            .get("payload")
22            .ok_or(NormalizationError::MissingField("payload"))?
23            .clone();
24        return enforce_source_consistency(EventEnvelope {
25            event_id,
26            source,
27            timestamp_ms,
28            event: UnifiedEvent::Module(ModuleEvent {
29                module,
30                event_type,
31                payload,
32            }),
33        });
34    }
35
36    let agent_id = required_string(object.get("agent_id"), "agent_id")?;
37    let event_type = required_string(object.get("event_type"), "event_type")?;
38    let payload = object.get("payload").cloned();
39
40    enforce_source_consistency(EventEnvelope {
41        event_id,
42        source,
43        timestamp_ms,
44        event: UnifiedEvent::Agent {
45            agent_id,
46            event_type,
47            payload,
48        },
49    })
50}
51
52impl MobkitRuntimeHandle {
53    pub(crate) fn append_normalized_event(
54        &mut self,
55        event: EventEnvelope<UnifiedEvent>,
56    ) -> Result<(), NormalizationError> {
57        let event = enforce_source_consistency(event)?;
58        insert_event_sorted(&mut self.merged_events, event);
59        Ok(())
60    }
61
62    pub fn merged_events(&self) -> &[EventEnvelope<UnifiedEvent>] {
63        &self.merged_events
64    }
65    pub fn subscribe_events(
66        &self,
67        request: SubscribeRequest,
68    ) -> Result<SubscribeResponse, SubscribeError> {
69        if let Some(checkpoint) = request.last_event_id.as_ref()
70            && checkpoint.trim().is_empty()
71        {
72            return Err(SubscribeError::EmptyCheckpoint);
73        }
74
75        if matches!(request.scope, SubscribeScope::Agent) {
76            let agent_id = request
77                .agent_id
78                .as_deref()
79                .ok_or(SubscribeError::MissingAgentId)?;
80            if agent_id.trim().is_empty() {
81                return Err(SubscribeError::InvalidAgentId);
82            }
83        }
84
85        let scoped_events: Vec<_> = self
86            .merged_events
87            .iter()
88            .filter(|event| event_matches_request(event, &request))
89            .collect();
90        let skip = scoped_events
91            .len()
92            .saturating_sub(SUBSCRIBE_REPLAY_EVENT_CAP);
93        let bounded = &scoped_events[skip..];
94
95        let replay_slice = match request.last_event_id.as_ref() {
96            Some(checkpoint) => {
97                let start_idx = bounded
98                    .iter()
99                    .position(|event| event.event_id == *checkpoint)
100                    .ok_or_else(|| SubscribeError::UnknownCheckpoint(checkpoint.clone()))?;
101                &bounded[start_idx..]
102            }
103            None => bounded,
104        };
105        let replay_events: Vec<_> = replay_slice.iter().map(|e| (*e).clone()).collect();
106        let event_frames = replay_events
107            .iter()
108            .map(build_sse_event_frame)
109            .collect::<Vec<_>>();
110
111        Ok(SubscribeResponse {
112            scope: request.scope,
113            replay_from_event_id: request.last_event_id,
114            keep_alive: SubscribeKeepAlive {
115                interval_ms: SSE_KEEP_ALIVE_INTERVAL_MS,
116                event: SSE_KEEP_ALIVE_EVENT_NAME.to_string(),
117            },
118            keep_alive_comment: SSE_KEEP_ALIVE_COMMENT_FRAME.to_string(),
119            event_frames,
120            events: replay_events,
121        })
122    }
123}
124
125pub(super) fn merge_unified_events(
126    mut module_events: Vec<EventEnvelope<UnifiedEvent>>,
127    mut agent_events: Vec<EventEnvelope<UnifiedEvent>>,
128) -> Vec<EventEnvelope<UnifiedEvent>> {
129    let mut merged = Vec::with_capacity(module_events.len() + agent_events.len());
130    merged.append(&mut module_events);
131    merged.append(&mut agent_events);
132    merged.sort_by(|left, right| {
133        left.timestamp_ms
134            .cmp(&right.timestamp_ms)
135            .then_with(|| left.event_id.cmp(&right.event_id))
136            .then_with(|| left.source.cmp(&right.source))
137    });
138    merged
139}
140
141/// True when a caller-supplied agent-scope filter matches a projected
142/// `agent_id`. The projected id is the generation-suffixed public alias
143/// (`{alias}:{generation}`, e.g. `worker:0` or `rt:review:singleton:0:1`), but
144/// callers naturally pass the BARE member id / durable identity (the value every
145/// sibling identity surface accepts). Accept either form: an exact match, or the
146/// bare alias whose only difference is a trailing `:{generation}` segment.
147fn agent_scope_filter_matches(selected: &str, agent_id: &str) -> bool {
148    if selected == agent_id {
149        return true;
150    }
151    // `agent_id == "{selected}:{generation}"` where generation is numeric.
152    agent_id
153        .strip_prefix(selected)
154        .and_then(|rest| rest.strip_prefix(':'))
155        .is_some_and(|generation| {
156            !generation.is_empty() && generation.bytes().all(|b| b.is_ascii_digit())
157        })
158}
159
160fn event_matches_request(event: &EventEnvelope<UnifiedEvent>, request: &SubscribeRequest) -> bool {
161    match request.scope {
162        SubscribeScope::Mob => true,
163        SubscribeScope::Agent => match &event.event {
164            UnifiedEvent::Agent { agent_id, .. } => request
165                .agent_id
166                .as_deref()
167                .map(|selected| agent_scope_filter_matches(selected, agent_id))
168                .unwrap_or(false),
169            UnifiedEvent::Module(_) => false,
170        },
171        SubscribeScope::Interaction => match &event.event {
172            UnifiedEvent::Agent { event_type, .. } => event_type.starts_with("interaction"),
173            UnifiedEvent::Module(module_event) => {
174                module_event.event_type.starts_with("interaction")
175            }
176        },
177    }
178}
179
180fn build_sse_event_frame(event: &EventEnvelope<UnifiedEvent>) -> String {
181    let event_name = match &event.event {
182        UnifiedEvent::Agent { event_type, .. } => event_type.as_str(),
183        UnifiedEvent::Module(module_event) => module_event.event_type.as_str(),
184    };
185    let payload = serde_json::to_string(&event.event).unwrap_or_else(|_| "{}".to_string());
186    format!(
187        "id: {}\nevent: {}\ndata: {}\n\n",
188        event.event_id, event_name, payload
189    )
190}
191
192fn enforce_source_consistency(
193    envelope: EventEnvelope<UnifiedEvent>,
194) -> Result<EventEnvelope<UnifiedEvent>, NormalizationError> {
195    let expected = match &envelope.event {
196        UnifiedEvent::Agent { .. } => "agent",
197        UnifiedEvent::Module(_) => "module",
198    };
199    if envelope.source != expected {
200        return Err(NormalizationError::SourceMismatch {
201            expected,
202            got: envelope.source,
203        });
204    }
205    Ok(envelope)
206}
207
208fn required_string(
209    value: Option<&Value>,
210    field: &'static str,
211) -> Result<String, NormalizationError> {
212    let value = value.ok_or(NormalizationError::MissingField(field))?;
213    let text = value
214        .as_str()
215        .ok_or(NormalizationError::InvalidFieldType(field))?;
216    Ok(text.to_string())
217}
218
219fn required_u64(value: Option<&Value>, field: &'static str) -> Result<u64, NormalizationError> {
220    let value = value.ok_or(NormalizationError::MissingField(field))?;
221    value
222        .as_u64()
223        .ok_or(NormalizationError::InvalidFieldType(field))
224}
225
226pub(super) fn insert_event_sorted(
227    events: &mut Vec<EventEnvelope<UnifiedEvent>>,
228    event: EventEnvelope<UnifiedEvent>,
229) {
230    let insertion_index = events
231        .binary_search_by(|existing| {
232            existing
233                .timestamp_ms
234                .cmp(&event.timestamp_ms)
235                .then_with(|| existing.event_id.cmp(&event.event_id))
236                .then_with(|| existing.source.cmp(&event.source))
237        })
238        .unwrap_or_else(|index| index);
239    events.insert(insertion_index, event);
240}
241
242#[cfg(test)]
243mod tests {
244    use super::agent_scope_filter_matches;
245
246    #[test]
247    fn agent_scope_filter_accepts_bare_and_generation_suffixed_ids() {
248        // Real events carry the generation-suffixed alias. A caller filtering by
249        // the bare member id / durable identity (what every sibling identity
250        // surface accepts) must match — the old exact-compare returned zero
251        // events silently. Regression for the agent-scope filter id mismatch.
252
253        // Bare alias matches the generation-suffixed projected id.
254        assert!(agent_scope_filter_matches("worker", "worker:0"));
255        assert!(agent_scope_filter_matches("worker", "worker:12"));
256        // Identity-first alias with embedded colons + a trailing generation.
257        assert!(agent_scope_filter_matches(
258            "rt:review:singleton:0",
259            "rt:review:singleton:0:1"
260        ));
261        // Exact match still works (caller passed the full suffixed id).
262        assert!(agent_scope_filter_matches("worker:0", "worker:0"));
263
264        // Must NOT over-match a different member that merely shares a prefix.
265        assert!(!agent_scope_filter_matches("worker", "worker-2:0"));
266        assert!(!agent_scope_filter_matches("work", "worker:0"));
267        // The trailing segment must be a numeric generation, not arbitrary text.
268        assert!(!agent_scope_filter_matches("worker", "worker:abc"));
269        assert!(!agent_scope_filter_matches("worker", "worker:"));
270    }
271}
meerkat_mobkit/runtime/event_transport.rs

meerkat_mobkit/runtime/
event_transport.rs
