agentmux_srv/agents/translator/
codex.rs

1// Copyright 2026, AgentMux Corp.
2// SPDX-License-Identifier: Apache-2.0
3
4//! Codex `exec --json` → `AgentEvent` translator — the answer text only.
5//!
6//! Scoped to what a tool-less, one-shot turn produces (`/btw`,
7//! `server/agent_handlers/side_question.rs`): assistant text, the end of
8//! the turn, and failures. Tool items (`command_execution`,
9//! `mcp_tool_call`, …) are dropped rather than translated; the full
10//! renderer is the frontend's `providers/codex-translator.ts`, which this
11//! follows for the frames it does handle:
12//!
13//! - `item.started` / `item.updated` / `item.completed` with
14//!   `item.type = "agent_message"` (text in `item.text`) or
15//!   `item.type = "message"`, `role = "assistant"` (text in
16//!   `output_text` content blocks). Each carries the item's text SO FAR,
17//!   not a delta, so only the new suffix is emitted as `AssistantText`.
18//! - `turn.completed` — `Done`, whose `response` is every message's text.
19//! - `turn.failed` / `error` — `Error`. An `error` whose message contains
20//!   `Reconnecting...` is a transient retry notice, not a failure.
21
22use std::collections::HashMap;
23
24use serde_json::Value;
25
26use super::super::types::AgentEvent;
27
28#[derive(Default)]
29pub struct CodexTranslator {
30    /// item id → text already emitted for it.
31    emitted: HashMap<String, String>,
32    /// Item ids in first-seen order, for assembling `Done.response`.
33    order: Vec<String>,
34    terminal: bool,
35}
36
37impl CodexTranslator {
38    pub fn new() -> Self {
39        Self::default()
40    }
41
42    fn item_text(item: &Value) -> Option<String> {
43        match item.get("type").and_then(Value::as_str)? {
44            "agent_message" => Some(item.get("text").and_then(Value::as_str).unwrap_or("").to_string()),
45            "message" if item.get("role").and_then(Value::as_str) == Some("assistant") => Some(
46                item.get("content")
47                    .and_then(Value::as_array)
48                    .map(|blocks| {
49                        blocks
50                            .iter()
51                            .filter(|b| b.get("type").and_then(Value::as_str) == Some("output_text"))
52                            .filter_map(|b| b.get("text").and_then(Value::as_str))
53                            .collect::<String>()
54                    })
55                    .unwrap_or_default(),
56            ),
57            _ => None,
58        }
59    }
60
61    fn translate_item(&mut self, item: &Value) -> Vec<AgentEvent> {
62        let Some(text) = Self::item_text(item) else {
63            return Vec::new();
64        };
65        let id = item
66            .get("id")
67            .and_then(Value::as_str)
68            .unwrap_or("message:anonymous")
69            .to_string();
70        if !self.emitted.contains_key(&id) {
71            self.order.push(id.clone());
72        }
73        let prev = self.emitted.entry(id).or_default();
74        match text.strip_prefix(prev.as_str()) {
75            Some("") => Vec::new(),
76            Some(suffix) => {
77                let delta = suffix.to_string();
78                *prev = text;
79                vec![AgentEvent::AssistantText { delta }]
80            }
81            // A snapshot that doesn't extend what was already shown (a
82            // rewrite) can't be a delta; `Done.response` carries the final
83            // text, which the overlay swaps in.
84            None => {
85                *prev = text;
86                Vec::new()
87            }
88        }
89    }
90
91    fn error_message(frame: &Value, fallback: &str) -> String {
92        let err = frame.get("error").unwrap_or(frame);
93        err.get("message")
94            .and_then(Value::as_str)
95            .or_else(|| err.as_str())
96            .unwrap_or(fallback)
97            .to_string()
98    }
99
100    fn finish(&mut self, event: AgentEvent) -> Vec<AgentEvent> {
101        if self.terminal {
102            return Vec::new();
103        }
104        self.terminal = true;
105        vec![event]
106    }
107}
108
109impl super::Translator for CodexTranslator {
110    fn translate(&mut self, frame: Value) -> Vec<AgentEvent> {
111        match frame.get("type").and_then(Value::as_str).unwrap_or("") {
112            "item.started" | "item.updated" | "item.completed" => match frame.get("item") {
113                Some(item) => self.translate_item(item),
114                None => Vec::new(),
115            },
116            "turn.completed" => {
117                let response = self
118                    .order
119                    .iter()
120                    .filter_map(|id| self.emitted.get(id))
121                    .filter(|t| !t.is_empty())
122                    .cloned()
123                    .collect::<Vec<_>>()
124                    .join("\n\n");
125                self.finish(AgentEvent::Done {
126                    response,
127                    transcript: Vec::new(),
128                })
129            }
130            "turn.failed" => {
131                let message = Self::error_message(&frame, "Codex turn failed");
132                self.finish(AgentEvent::Error { message })
133            }
134            "error" => {
135                let message = Self::error_message(&frame, "Codex error");
136                if message.contains("Reconnecting...") {
137                    return Vec::new();
138                }
139                self.finish(AgentEvent::Error { message })
140            }
141            _ => Vec::new(),
142        }
143    }
144}
145
146#[cfg(test)]
147mod tests {
148    use super::super::Translator;
149    use super::*;
150    use serde_json::json;
151
152    fn texts(events: &[AgentEvent]) -> Vec<String> {
153        events
154            .iter()
155            .filter_map(|e| match e {
156                AgentEvent::AssistantText { delta } => Some(delta.clone()),
157                _ => None,
158            })
159            .collect()
160    }
161
162    #[test]
163    fn snapshots_become_deltas_and_the_turn_end_is_done() {
164        let mut t = CodexTranslator::new();
165        let mut out = Vec::new();
166        for frame in [
167            json!({"type": "thread.started", "thread_id": "th-1"}),
168            json!({"type": "turn.started"}),
169            json!({"type": "item.started", "item": {"id": "m1", "type": "agent_message", "text": "Hel"}}),
170            json!({"type": "item.updated", "item": {"id": "m1", "type": "agent_message", "text": "Hello"}}),
171            json!({"type": "item.completed", "item": {"id": "m1", "type": "agent_message", "text": "Hello."}}),
172            json!({"type": "turn.completed", "usage": {"input_tokens": 3, "output_tokens": 2}}),
173        ] {
174            out.extend(t.translate(frame));
175        }
176        assert_eq!(texts(&out), ["Hel", "lo", "."]);
177        match out.last() {
178            Some(AgentEvent::Done { response, .. }) => assert_eq!(response, "Hello."),
179            other => panic!("expected Done, got {other:?}"),
180        }
181    }
182
183    #[test]
184    fn assistant_message_items_carry_output_text() {
185        let mut t = CodexTranslator::new();
186        let out = t.translate(json!({"type": "item.completed", "item": {
187            "id": "m1", "type": "message", "role": "assistant",
188            "content": [{"type": "output_text", "text": "Hi"}, {"type": "other", "text": "x"}]
189        }}));
190        assert_eq!(texts(&out), ["Hi"]);
191    }
192
193    #[test]
194    fn tool_items_and_user_messages_are_dropped() {
195        let mut t = CodexTranslator::new();
196        assert!(t
197            .translate(json!({"type": "item.started", "item": {"id": "c1", "type": "command_execution", "command": "ls"}}))
198            .is_empty());
199        assert!(t
200            .translate(json!({"type": "item.completed", "item": {"id": "u1", "type": "message", "role": "user", "content": []}}))
201            .is_empty());
202    }
203
204    #[test]
205    fn a_failed_turn_is_one_error_and_reconnects_are_not_failures() {
206        let mut t = CodexTranslator::new();
207        assert!(t.translate(json!({"type": "error", "message": "Reconnecting... 1/5"})).is_empty());
208        let out = t.translate(json!({"type": "turn.failed", "error": {"message": "quota exceeded"}}));
209        match out.as_slice() {
210            [AgentEvent::Error { message }] => assert_eq!(message, "quota exceeded"),
211            other => panic!("expected one Error, got {other:?}"),
212        }
213        assert!(t.translate(json!({"type": "error", "message": "late"})).is_empty(), "only one terminal event");
214    }
215}
agentmux_srv/agents/translator/codex.rs

agentmux_srv/agents/translator/
codex.rs
