Skip to main content

mempal_runtime/ingest/
normalize.rs

1use serde_json::Value;
2use thiserror::Error;
3
4use super::detect::{Format, extract_content_text, extract_message_text};
5use super::noise::{strip_claude_jsonl_noise, strip_codex_rollout_noise};
6
7pub const CURRENT_NORMALIZE_VERSION: u32 = 3;
8
9pub type Result<T> = std::result::Result<T, NormalizeError>;
10
11#[derive(Debug, Clone, Copy, PartialEq, Eq)]
12pub struct NormalizeOptions {
13    pub strip_noise: bool,
14}
15
16impl Default for NormalizeOptions {
17    fn default() -> Self {
18        Self { strip_noise: true }
19    }
20}
21
22#[derive(Debug, Clone, PartialEq, Eq)]
23pub struct NormalizeOutput {
24    pub content: String,
25    pub noise_bytes_stripped: Option<u64>,
26}
27
28#[derive(Debug, Error)]
29pub enum NormalizeError {
30    #[error(transparent)]
31    Json(#[from] serde_json::Error),
32    #[error("unsupported ChatGPT JSON shape")]
33    UnsupportedChatGptShape,
34}
35
36pub fn normalize_content(content: &str, format: Format) -> Result<String> {
37    Ok(normalize_content_with_options(content, format, NormalizeOptions::default())?.content)
38}
39
40pub fn normalize_content_with_options(
41    content: &str,
42    format: Format,
43    options: NormalizeOptions,
44) -> Result<NormalizeOutput> {
45    match format {
46        Format::PlainText => Ok(NormalizeOutput {
47            content: content.trim().to_string(),
48            noise_bytes_stripped: None,
49        }),
50        Format::ClaudeJsonl => normalize_claude_jsonl(content, options.strip_noise),
51        Format::ChatGptJson => Ok(NormalizeOutput {
52            content: normalize_chatgpt_json(content)?,
53            noise_bytes_stripped: None,
54        }),
55        Format::CodexJsonl => normalize_codex_jsonl(content, options.strip_noise),
56        Format::SlackJson => Ok(NormalizeOutput {
57            content: normalize_slack_json(content)?,
58            noise_bytes_stripped: None,
59        }),
60    }
61}
62
63fn normalize_claude_jsonl(content: &str, strip_noise: bool) -> Result<NormalizeOutput> {
64    let mut lines = Vec::new();
65    let mut noise_bytes_stripped = 0_u64;
66
67    for raw_line in content
68        .lines()
69        .map(str::trim)
70        .filter(|line| !line.is_empty())
71    {
72        let value: Value = serde_json::from_str(raw_line)?;
73        let role = value
74            .get("type")
75            .and_then(Value::as_str)
76            .unwrap_or("assistant");
77        let message = extract_message_text(&value).unwrap_or_default();
78        let message = message.trim();
79        let message = if strip_noise {
80            let stripped = strip_claude_jsonl_noise(message);
81            noise_bytes_stripped += message.len().saturating_sub(stripped.len()) as u64;
82            stripped
83        } else {
84            message.to_string()
85        };
86
87        if message.trim().is_empty() {
88            continue;
89        }
90
91        if matches!(role, "human" | "user") {
92            lines.push(format!("> {}", message.trim()));
93        } else {
94            lines.push(message.trim().to_string());
95        }
96    }
97
98    Ok(NormalizeOutput {
99        content: lines.join("\n"),
100        noise_bytes_stripped: strip_noise.then_some(noise_bytes_stripped),
101    })
102}
103
104fn normalize_chatgpt_json(content: &str) -> Result<String> {
105    let value: Value = serde_json::from_str(content)?;
106
107    if let Some(messages) = value.as_array() {
108        return normalize_chatgpt_messages(messages);
109    }
110
111    if let Some(messages) = value.get("messages").and_then(Value::as_array) {
112        return normalize_chatgpt_messages(messages);
113    }
114
115    if let Some(mapping) = value.get("mapping").and_then(Value::as_object) {
116        let mut ordered = Vec::new();
117        if let Some(root_id) = find_root_node(mapping) {
118            collect_messages_dfs(mapping, &root_id, &mut ordered);
119        }
120
121        return Ok(render_transcript(ordered));
122    }
123
124    Err(NormalizeError::UnsupportedChatGptShape)
125}
126
127fn normalize_chatgpt_messages(messages: &[Value]) -> Result<String> {
128    let transcript = render_transcript(messages.iter().filter_map(|message| {
129        let role = message.get("role").and_then(Value::as_str)?;
130        let content = message.get("content").and_then(extract_content_text)?;
131        Some((role.to_string(), content))
132    }));
133
134    Ok(transcript)
135}
136
137fn find_root_node(mapping: &serde_json::Map<String, Value>) -> Option<String> {
138    mapping
139        .iter()
140        .find(|(_, node)| {
141            node.get("parent")
142                .is_none_or(|p| p.is_null() || p.as_str() == Some(""))
143        })
144        .map(|(id, _)| id.clone())
145}
146
147fn collect_messages_dfs(
148    mapping: &serde_json::Map<String, Value>,
149    node_id: &str,
150    result: &mut Vec<(String, String)>,
151) {
152    let Some(node) = mapping.get(node_id) else {
153        return;
154    };
155
156    if let Some(message) = node.get("message") {
157        let role = message
158            .get("author")
159            .and_then(|author| author.get("role"))
160            .and_then(Value::as_str);
161        let content = message.get("content").and_then(extract_content_text);
162        if let (Some(role), Some(content)) = (role, content) {
163            result.push((role.to_string(), content));
164        }
165    }
166
167    if let Some(children) = node.get("children").and_then(Value::as_array) {
168        for child in children {
169            if let Some(child_id) = child.as_str() {
170                collect_messages_dfs(mapping, child_id, result);
171            }
172        }
173    }
174}
175
176fn normalize_codex_jsonl(content: &str, strip_noise: bool) -> Result<NormalizeOutput> {
177    let mut response_items: Vec<(String, String)> = Vec::new();
178    let mut legacy_events: Vec<(String, String)> = Vec::new();
179
180    for line in content.lines().map(str::trim).filter(|l| !l.is_empty()) {
181        let value: Value = serde_json::from_str(line)?;
182        let record_type = value.get("type").and_then(Value::as_str).unwrap_or("");
183        let Some(payload) = value.get("payload") else {
184            continue;
185        };
186        match record_type {
187            "response_item" => {
188                if payload.get("type").and_then(Value::as_str) != Some("message") {
189                    continue;
190                }
191
192                let role = payload.get("role").and_then(Value::as_str).unwrap_or("");
193                if role != "user" && role != "assistant" {
194                    continue;
195                }
196
197                let Some(message) = payload.get("content").and_then(extract_content_text) else {
198                    continue;
199                };
200                let message = message.trim();
201                if message.is_empty() {
202                    continue;
203                }
204
205                response_items.push((role.to_string(), message.to_string()));
206            }
207            "event_msg" => {
208                let msg_type = payload.get("type").and_then(Value::as_str).unwrap_or("");
209                let message = payload
210                    .get("message")
211                    .and_then(Value::as_str)
212                    .unwrap_or("")
213                    .trim();
214                if message.is_empty() {
215                    continue;
216                }
217
218                match msg_type {
219                    "user_message" => legacy_events.push(("user".to_string(), message.to_string())),
220                    "agent_message" => {
221                        legacy_events.push(("assistant".to_string(), message.to_string()))
222                    }
223                    _ => {}
224                }
225            }
226            _ => {}
227        }
228    }
229
230    let mut pairs = if response_items.is_empty() {
231        legacy_events
232    } else {
233        response_items
234    };
235    let mut noise_bytes_stripped = 0_u64;
236    if strip_noise {
237        for (_, message) in &mut pairs {
238            let stripped = strip_codex_rollout_noise(message);
239            noise_bytes_stripped += message.len().saturating_sub(stripped.len()) as u64;
240            *message = stripped;
241        }
242    }
243
244    Ok(NormalizeOutput {
245        content: render_transcript(pairs),
246        noise_bytes_stripped: strip_noise.then_some(noise_bytes_stripped),
247    })
248}
249
250fn normalize_slack_json(content: &str) -> Result<String> {
251    let value: Value = serde_json::from_str(content)?;
252    let messages = value
253        .as_array()
254        .ok_or(NormalizeError::UnsupportedChatGptShape)?;
255
256    let mut speakers: Vec<String> = Vec::new();
257    let mut pairs: Vec<(String, String)> = Vec::new();
258
259    for msg in messages {
260        if msg.get("type").and_then(Value::as_str) != Some("message") {
261            continue;
262        }
263        let speaker = msg
264            .get("user")
265            .or_else(|| msg.get("username"))
266            .and_then(Value::as_str)
267            .unwrap_or("unknown")
268            .to_string();
269        let text = msg.get("text").and_then(Value::as_str).unwrap_or("").trim();
270        if text.is_empty() {
271            continue;
272        }
273
274        // First speaker = user, second = assistant
275        if !speakers.contains(&speaker) {
276            speakers.push(speaker.clone());
277        }
278        let role = if speakers.first() == Some(&speaker) {
279            "user"
280        } else {
281            "assistant"
282        };
283        pairs.push((role.to_string(), text.to_string()));
284    }
285
286    Ok(render_transcript(pairs))
287}
288
289fn render_transcript(items: impl IntoIterator<Item = (String, String)>) -> String {
290    let mut lines = Vec::new();
291
292    for (role, content) in items {
293        if content.trim().is_empty() {
294            continue;
295        }
296
297        if matches!(role.as_str(), "user" | "human") {
298            lines.push(format!("> {}", content.trim()));
299        } else {
300            lines.push(content.trim().to_string());
301        }
302    }
303
304    lines.join("\n")
305}
306
307#[cfg(test)]
308mod tests {
309    use super::normalize_codex_jsonl;
310
311    #[test]
312    fn codex_normalize_prefers_response_item_messages() {
313        let content = r#"{"timestamp":"2026-04-19T10:37:36.000Z","type":"session_meta","payload":{"cwd":"/tmp/project"}}
314{"timestamp":"2026-04-19T10:37:36.050Z","type":"response_item","payload":{"type":"message","role":"developer","content":[{"type":"input_text","text":"developer instructions"}]}}
315{"timestamp":"2026-04-19T10:37:36.100Z","type":"response_item","payload":{"type":"message","role":"user","content":[{"type":"input_text","text":"first line"},{"type":"input_text","text":"second line"}]}}
316{"timestamp":"2026-04-19T10:37:36.150Z","type":"event_msg","payload":{"type":"user_message","message":"duplicate legacy user"}}
317{"timestamp":"2026-04-19T10:37:36.200Z","type":"response_item","payload":{"type":"message","role":"assistant","content":[{"type":"output_text","text":"answer"}]}}
318{"timestamp":"2026-04-19T10:37:36.250Z","type":"event_msg","payload":{"type":"agent_message","message":"duplicate legacy assistant"}}
319{"timestamp":"2026-04-19T10:37:36.300Z","type":"compacted","payload":{"summary":"trimmed"}}"#;
320
321        let normalized = normalize_codex_jsonl(content, true).expect("normalize codex");
322        assert_eq!(normalized.content, "> first line\nsecond line\nanswer");
323    }
324
325    #[test]
326    fn codex_normalize_drops_runtime_preamble_user_messages() {
327        // Issue #10: Codex injects runtime preamble as role=user messages;
328        // wrapper-only messages must not become transcript lines.
329        let content = r#"{"timestamp":"2026-07-26T10:00:00.000Z","type":"session_meta","payload":{"cwd":"/tmp/project"}}
330{"timestamp":"2026-07-26T10:00:00.100Z","type":"response_item","payload":{"type":"message","role":"user","content":[{"type":"input_text","text":"<user_instructions>\nAGENTS.md content\n</user_instructions>"}]}}
331{"timestamp":"2026-07-26T10:00:00.200Z","type":"response_item","payload":{"type":"message","role":"user","content":[{"type":"input_text","text":"<environment_context>\n<cwd>/tmp/project</cwd>\n</environment_context>\nfix the bug"}]}}
332{"timestamp":"2026-07-26T10:00:00.300Z","type":"response_item","payload":{"type":"message","role":"assistant","content":[{"type":"output_text","text":"done"}]}}"#;
333
334        let normalized = normalize_codex_jsonl(content, true).expect("normalize codex");
335        assert_eq!(normalized.content, "> fix the bug\ndone");
336    }
337
338    #[test]
339    fn codex_normalize_falls_back_to_legacy_event_messages() {
340        let content = r#"{"timestamp":"2026-04-13T12:00:00Z","type":"session_meta","payload":{"cwd":"/tmp/project"}}
341{"timestamp":"2026-04-13T12:00:10Z","type":"event_msg","payload":{"type":"user_message","message":"legacy hello"}}
342{"timestamp":"2026-04-13T12:00:20Z","type":"event_msg","payload":{"type":"agent_message","message":"legacy hi"}}"#;
343
344        let normalized = normalize_codex_jsonl(content, true).expect("normalize codex");
345        assert_eq!(normalized.content, "> legacy hello\nlegacy hi");
346    }
347}