mempal_runtime/ingest/
detect.rs

1use serde_json::Value;
2
3#[derive(Debug, Clone, Copy, PartialEq, Eq)]
4pub enum Format {
5    ClaudeJsonl,
6    ChatGptJson,
7    CodexJsonl,
8    SlackJson,
9    PlainText,
10}
11
12pub fn detect_format(content: &str) -> Format {
13    if is_claude_jsonl(content) {
14        return Format::ClaudeJsonl;
15    }
16
17    if is_codex_jsonl(content) {
18        return Format::CodexJsonl;
19    }
20
21    if is_slack_json(content) {
22        return Format::SlackJson;
23    }
24
25    if is_chatgpt_json(content) {
26        return Format::ChatGptJson;
27    }
28
29    Format::PlainText
30}
31
32fn is_codex_jsonl(content: &str) -> bool {
33    let mut has_session_meta = false;
34    let mut has_message_record = false;
35
36    for line in content.lines().map(str::trim).filter(|l| !l.is_empty()) {
37        let Ok(value) = serde_json::from_str::<Value>(line) else {
38            return false;
39        };
40        match value.get("type").and_then(Value::as_str) {
41            Some("session_meta") => has_session_meta = true,
42            Some("event_msg" | "response_item") => has_message_record = true,
43            // Tolerate turn_context/compacted and newer rollout record types,
44            // but they are not evidence of a Codex rollout on their own.
45            Some(_) => {}
46            None => return false,
47        }
48    }
49
50    has_session_meta && has_message_record
51}
52
53fn is_slack_json(content: &str) -> bool {
54    let Ok(value) = serde_json::from_str::<Value>(content) else {
55        return false;
56    };
57    let Some(arr) = value.as_array() else {
58        return false;
59    };
60    // Slack messages have "type": "message" and "user"/"username" + "text"
61    arr.iter().take(5).any(|msg| {
62        msg.get("type").and_then(Value::as_str) == Some("message")
63            && (msg.get("user").is_some() || msg.get("username").is_some())
64            && msg.get("text").is_some()
65    })
66}
67
68fn is_claude_jsonl(content: &str) -> bool {
69    let mut saw_line = false;
70
71    for line in content
72        .lines()
73        .map(str::trim)
74        .filter(|line| !line.is_empty())
75    {
76        let Ok(value) = serde_json::from_str::<Value>(line) else {
77            return false;
78        };
79
80        if value.get("type").and_then(Value::as_str).is_none() {
81            return false;
82        }
83        if extract_message_text(&value).is_none() {
84            return false;
85        }
86
87        saw_line = true;
88    }
89
90    saw_line
91}
92
93fn is_chatgpt_json(content: &str) -> bool {
94    let Ok(value) = serde_json::from_str::<Value>(content) else {
95        return false;
96    };
97
98    matches!(value, Value::Array(_))
99        || value.get("messages").is_some()
100        || value.get("mapping").is_some()
101}
102
103pub(crate) fn extract_message_text(value: &Value) -> Option<String> {
104    value
105        .get("message")
106        .and_then(Value::as_str)
107        .map(ToOwned::to_owned)
108        .or_else(|| value.get("content").and_then(extract_content_text))
109}
110
111pub(crate) fn extract_content_text(value: &Value) -> Option<String> {
112    match value {
113        Value::String(text) => Some(text.clone()),
114        Value::Array(items) => Some(
115            items
116                .iter()
117                .filter_map(|item| {
118                    item.as_str().map(ToOwned::to_owned).or_else(|| {
119                        item.get("text")
120                            .and_then(Value::as_str)
121                            .map(ToOwned::to_owned)
122                    })
123                })
124                .collect::<Vec<_>>()
125                .join("\n"),
126        ),
127        Value::Object(map) => map
128            .get("parts")
129            .and_then(Value::as_array)
130            .map(|parts| {
131                parts
132                    .iter()
133                    .filter_map(Value::as_str)
134                    .collect::<Vec<_>>()
135                    .join("\n")
136            })
137            .filter(|text| !text.is_empty()),
138        _ => None,
139    }
140}
141
142#[cfg(test)]
143mod tests {
144    use super::{Format, detect_format};
145
146    #[test]
147    fn rejects_codex_rollout_without_message_records() {
148        // session_meta plus non-message records only: turn_context/compacted
149        // are tolerated context, not sufficient evidence of a Codex rollout.
150        let content = r#"{"timestamp":"2026-07-26T10:00:00.000Z","type":"session_meta","payload":{"cwd":"/tmp/project"}}
151{"timestamp":"2026-07-26T10:00:00.100Z","type":"turn_context","payload":{"cwd":"/tmp/project"}}
152{"timestamp":"2026-07-26T10:00:00.200Z","type":"compacted","payload":{"summary":"trimmed"}}"#;
153
154        assert_eq!(detect_format(content), Format::PlainText);
155    }
156
157    #[test]
158    fn detects_current_codex_rollout_with_turn_context_and_compacted() {
159        let content = r#"{"timestamp":"2026-04-19T10:37:36.000Z","type":"session_meta","payload":{"cwd":"/tmp/project"}}
160{"timestamp":"2026-04-19T10:37:36.100Z","type":"turn_context","payload":{"cwd":"/tmp/project"}}
161{"timestamp":"2026-04-19T10:37:36.200Z","type":"response_item","payload":{"type":"message","role":"developer","content":[{"type":"input_text","text":"instructions"}]}}
162{"timestamp":"2026-04-19T10:37:36.300Z","type":"compacted","payload":{"summary":"trimmed"}}
163{"timestamp":"2026-04-19T10:37:36.400Z","type":"event_msg","payload":{"type":"token_count"}}"#;
164
165        assert_eq!(detect_format(content), Format::CodexJsonl);
166    }
167}
mempal_runtime/ingest/detect.rs

mempal_runtime/ingest/
detect.rs
