aion_integration_cli/
demux.rs

1//! The stdout demux — the pure translation from a plain CLI agent's interleaved stdout lines into
2//! neutral [`ActivityEvent`]s.
3//!
4//! This is the case-(b) analogue of the Norn adapter's `translate` module, and the whole point of
5//! the observability-only shape: a plain-stdout agent has **no** structured channel, so its
6//! transcript is recovered by parsing the interleaved lines it prints. The demux is deliberately
7//! **lossless**: any line the classifier does not recognise falls through to
8//! [`ActivityEventKind::Raw`] carrying the verbatim line, so nothing an agent prints is ever
9//! silently dropped.
10//!
11//! # The recognised shape
12//!
13//! A line that parses as a JSON object with a `"type"` string is mapped onto a neutral kind:
14//!
15//! | line `type` | neutral kind |
16//! |---|---|
17//! | `"message"` (with `"text"`) | [`ActivityEventKind::Message`] (assistant role) |
18//! | `"tool_call"` | [`ActivityEventKind::ToolCall`] |
19//! | `"tool_result"` | [`ActivityEventKind::ToolResult`] |
20//! | `"stop"` (with `"reason"`) | [`ActivityEventKind::Stop`] |
21//!
22//! Everything else — a non-JSON log line, a JSON value without a known `type`, an empty line — is a
23//! [`ActivityEventKind::Raw`]. This mirrors real observability-only CLIs, which interleave a little
24//! structured output with free-form logging.
25//!
26//! The demux is pure and independently tested; the session ([`crate::session`]) is the only place
27//! that reads the child's stdout and stamps ordering onto the events this produces.
28
29use aion_core::{ActivityEvent, ActivityEventKind, MessageRole, StopKind};
30use chrono::Utc;
31use serde_json::Value;
32use uuid::Uuid;
33
34/// The `(workflow, run, activity, attempt)` identity stamped onto every demuxed event.
35///
36/// The plain-stdout agent carries no agent sub-identity, so [`ActivityEvent::agent_id`] is always
37/// the nil UUID and [`ActivityEvent::agent_role`] is a fixed neutral label — an honest reflection
38/// of a single-agent CLI with no multi-agent attribution to recover.
39#[derive(Clone, Debug)]
40pub struct EventIdentity {
41    /// The workflow this run belongs to.
42    pub workflow_id: aion_core::WorkflowId,
43    /// The concrete run of that workflow this attempt belongs to — stamped from the run spec,
44    /// never synthesised here. It is the axis that keeps two generations of a continue-as-new
45    /// chain from writing one fused transcript.
46    pub run_id: aion_core::RunId,
47    /// The activity within the workflow.
48    pub activity_id: aion_core::ActivityId,
49    /// The attempt number of the run.
50    pub attempt: u32,
51}
52
53/// The fixed agent role for a single-agent plain-stdout CLI (no sub-agent attribution to recover).
54const CLI_AGENT_ROLE: &str = "cli";
55
56/// The `source` label a [`ActivityEventKind::Raw`] passthrough carries — the adapter's stdout.
57const RAW_SOURCE: &str = "stdout";
58
59/// Maps one line of the child's stdout into a neutral [`ActivityEvent`].
60///
61/// `worker_seq` is the session's best-effort local monotonic order for the line. A line that parses
62/// as a recognised JSON object maps onto the matching kind; anything else falls through to
63/// [`ActivityEventKind::Raw`], so no line is ever dropped.
64#[must_use]
65pub fn line_to_event(identity: &EventIdentity, worker_seq: u64, line: &str) -> ActivityEvent {
66    let kind = classify_line(line);
67    ActivityEvent {
68        workflow_id: identity.workflow_id.clone(),
69        run_id: identity.run_id.clone(),
70        activity_id: identity.activity_id.clone(),
71        attempt: identity.attempt,
72        agent_id: Uuid::nil(),
73        agent_role: CLI_AGENT_ROLE.to_owned(),
74        emitted_at: Utc::now(),
75        worker_seq,
76        store_seq: None,
77        // A plain-stdout CLI has no token-delta channel, so no event is ever ephemeral.
78        ephemeral: false,
79        kind,
80    }
81}
82
83/// Classifies a stdout line into an [`ActivityEventKind`].
84///
85/// A JSON object with a known `"type"` maps onto the matching kind; every other line (non-JSON, an
86/// unknown `type`, a bare value) is a [`ActivityEventKind::Raw`] carrying the line verbatim.
87fn classify_line(line: &str) -> ActivityEventKind {
88    let Some(object) = serde_json::from_str::<Value>(line)
89        .ok()
90        .filter(Value::is_object)
91    else {
92        return raw_kind(line);
93    };
94    match object.get("type").and_then(Value::as_str) {
95        Some("message") => message_kind(&object),
96        Some("tool_call") => tool_call_kind(&object),
97        Some("tool_result") => tool_result_kind(&object),
98        Some("stop") => stop_kind(&object),
99        // A JSON object whose `type` this adapter does not map is passed through verbatim, never
100        // dropped: the raw value is the parsed object so the console still sees the structure.
101        _ => ActivityEventKind::Raw {
102            source: RAW_SOURCE.to_owned(),
103            value: object,
104        },
105    }
106}
107
108/// A [`ActivityEventKind::Raw`] carrying a stdout line verbatim as a JSON string value.
109fn raw_kind(line: &str) -> ActivityEventKind {
110    ActivityEventKind::Raw {
111        source: RAW_SOURCE.to_owned(),
112        value: Value::String(line.to_owned()),
113    }
114}
115
116/// Reads a string field from a JSON object, or an empty string when absent.
117fn string_field(object: &Value, key: &str) -> String {
118    object
119        .get(key)
120        .and_then(Value::as_str)
121        .unwrap_or_default()
122        .to_owned()
123}
124
125/// Builds a [`ActivityEventKind::Message`] from a `{"type":"message","text":...}` line.
126///
127/// A plain-stdout CLI's structured messages are agent output, so the neutral role is always
128/// [`MessageRole::Assistant`].
129fn message_kind(object: &Value) -> ActivityEventKind {
130    ActivityEventKind::Message {
131        role: MessageRole::Assistant,
132        text: string_field(object, "text"),
133    }
134}
135
136/// Builds a [`ActivityEventKind::ToolCall`] from a `{"type":"tool_call",...}` line.
137fn tool_call_kind(object: &Value) -> ActivityEventKind {
138    ActivityEventKind::ToolCall {
139        tool: string_field(object, "tool"),
140        call_id: string_field(object, "call_id"),
141        input: object.get("input").cloned().unwrap_or(Value::Null),
142    }
143}
144
145/// Builds a [`ActivityEventKind::ToolResult`] from a `{"type":"tool_result",...}` line.
146fn tool_result_kind(object: &Value) -> ActivityEventKind {
147    ActivityEventKind::ToolResult {
148        call_id: string_field(object, "call_id"),
149        output: object.get("output").cloned().unwrap_or(Value::Null),
150        is_error: object
151            .get("is_error")
152            .and_then(Value::as_bool)
153            .unwrap_or(false),
154    }
155}
156
157/// Builds a [`ActivityEventKind::Stop`] from a `{"type":"stop","reason":...}` line.
158fn stop_kind(object: &Value) -> ActivityEventKind {
159    let reason = match object.get("reason").and_then(Value::as_str) {
160        Some("end_turn") => StopKind::EndTurn,
161        Some("tool_use") => StopKind::ToolUse,
162        Some("limit_reached") => StopKind::LimitReached,
163        Some("cancelled") => StopKind::Cancelled,
164        Some(other) => StopKind::Other {
165            reason: other.to_owned(),
166        },
167        None => StopKind::Other {
168            reason: "unknown".to_owned(),
169        },
170    };
171    ActivityEventKind::Stop { reason }
172}
173
174#[cfg(test)]
175#[allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)]
176mod tests {
177    use super::*;
178    use aion_core::{ActivityId, WorkflowId};
179    use serde_json::json;
180
181    fn identity() -> EventIdentity {
182        EventIdentity {
183            workflow_id: WorkflowId::new(Uuid::nil()),
184            run_id: aion_core::RunId::new(Uuid::from_u128(0x11)),
185            activity_id: ActivityId::from_sequence_position(5),
186            attempt: 2,
187        }
188    }
189
190    #[test]
191    fn message_line_maps_to_message_event() {
192        let line = json!({ "type": "message", "text": "working on it" }).to_string();
193        let event = line_to_event(&identity(), 0, &line);
194        assert_eq!(event.attempt, 2);
195        assert_eq!(event.agent_role, "cli");
196        assert_eq!(event.agent_id, Uuid::nil());
197        assert!(!event.ephemeral);
198        match event.kind {
199            ActivityEventKind::Message { role, text } => {
200                assert_eq!(role, MessageRole::Assistant);
201                assert_eq!(text, "working on it");
202            }
203            other => panic!("expected Message, got {other:?}"),
204        }
205    }
206
207    #[test]
208    fn tool_call_line_maps_to_tool_call_event() {
209        let line = json!({
210            "type": "tool_call",
211            "tool": "search",
212            "call_id": "c1",
213            "input": { "query": "x" },
214        })
215        .to_string();
216        let event = line_to_event(&identity(), 1, &line);
217        match event.kind {
218            ActivityEventKind::ToolCall {
219                tool,
220                call_id,
221                input,
222            } => {
223                assert_eq!(tool, "search");
224                assert_eq!(call_id, "c1");
225                assert_eq!(input, json!({ "query": "x" }));
226            }
227            other => panic!("expected ToolCall, got {other:?}"),
228        }
229    }
230
231    #[test]
232    fn tool_result_line_maps_to_tool_result_event() {
233        let line = json!({
234            "type": "tool_result",
235            "call_id": "c1",
236            "output": { "hits": 3 },
237            "is_error": false,
238        })
239        .to_string();
240        let event = line_to_event(&identity(), 2, &line);
241        match event.kind {
242            ActivityEventKind::ToolResult {
243                call_id,
244                output,
245                is_error,
246            } => {
247                assert_eq!(call_id, "c1");
248                assert_eq!(output, json!({ "hits": 3 }));
249                assert!(!is_error);
250            }
251            other => panic!("expected ToolResult, got {other:?}"),
252        }
253    }
254
255    #[test]
256    fn stop_line_maps_stop_reasons() {
257        for (label, expected) in [
258            ("end_turn", StopKind::EndTurn),
259            ("tool_use", StopKind::ToolUse),
260            ("limit_reached", StopKind::LimitReached),
261            ("cancelled", StopKind::Cancelled),
262        ] {
263            let line = json!({ "type": "stop", "reason": label }).to_string();
264            let event = line_to_event(&identity(), 3, &line);
265            match event.kind {
266                ActivityEventKind::Stop { reason } => assert_eq!(reason, expected),
267                other => panic!("expected Stop, got {other:?}"),
268            }
269        }
270    }
271
272    #[test]
273    fn unknown_stop_reason_falls_through_to_other() {
274        let line = json!({ "type": "stop", "reason": "weird" }).to_string();
275        let event = line_to_event(&identity(), 4, &line);
276        match event.kind {
277            ActivityEventKind::Stop {
278                reason: StopKind::Other { reason },
279            } => assert_eq!(reason, "weird"),
280            other => panic!("expected Stop::Other, got {other:?}"),
281        }
282    }
283
284    #[test]
285    fn a_plain_log_line_falls_through_to_raw_verbatim() {
286        let line = "  [info] starting the run, reading config...  ";
287        let event = line_to_event(&identity(), 5, line);
288        match event.kind {
289            ActivityEventKind::Raw { source, value } => {
290                assert_eq!(source, "stdout");
291                assert_eq!(value, Value::String(line.to_owned()));
292            }
293            other => panic!("expected Raw, got {other:?}"),
294        }
295    }
296
297    #[test]
298    fn a_json_object_with_unknown_type_is_raw_but_keeps_structure() {
299        let line = json!({ "type": "heartbeat", "seq": 4 }).to_string();
300        let event = line_to_event(&identity(), 6, &line);
301        match event.kind {
302            ActivityEventKind::Raw { source, value } => {
303                assert_eq!(source, "stdout");
304                assert_eq!(value["type"], json!("heartbeat"));
305                assert_eq!(value["seq"], json!(4));
306            }
307            other => panic!("expected Raw, got {other:?}"),
308        }
309    }
310
311    #[test]
312    fn a_bare_json_value_is_raw_not_mapped() {
313        // A JSON number/array/string is not an object, so it is Raw (a string of the verbatim line).
314        let line = "[1, 2, 3]";
315        let event = line_to_event(&identity(), 7, line);
316        match event.kind {
317            ActivityEventKind::Raw { value, .. } => {
318                assert_eq!(value, Value::String(line.to_owned()));
319            }
320            other => panic!("expected Raw, got {other:?}"),
321        }
322    }
323
324    #[test]
325    fn worker_seq_is_carried_onto_the_event() {
326        let event = line_to_event(&identity(), 42, "anything");
327        assert_eq!(event.worker_seq, 42);
328    }
329}
aion_integration_cli/demux.rs

aion_integration_cli/
demux.rs
