supercode_interchange/session/

codex.rs

//! Codex session codec: loaders, writers and native-record helpers.

use super::*;

impl Session {
    /// Load a Codex rollout from a file.
    pub fn from_codex(path: impl AsRef<Path>) -> Result<Session> {
        Self::from_codex_str(&std::fs::read_to_string(path.as_ref())?)
    }

    /// Parse a Codex rollout from an in-memory JSONL string.
    pub fn from_codex_str(jsonl: &str) -> Result<Session> {
        let mut meta = SessionMeta::new(SessionSource::Codex);
        let mut messages = Vec::new();

        // First pass: collect the text of every assistant message that exists as
        // a canonical `response_item`. In normal sessions the streamed
        // `event_msg/agent_message` events duplicate these and are safely
        // skipped; in collab/multi-agent sessions the assistant narration lives
        // ONLY as `agent_message` events, so we recover the ones with no
        // response_item counterpart (deduping by exact text).
        let assistant_texts = collect_codex_assistant_texts(jsonl);
        // IX-1: strict-verbatim raw capture — see `from_claude_code_str`.
        let (raw_lines, raw_trailing_newline) = split_lines_verbatim(jsonl);
        let raw: Vec<String> = raw_lines.iter().map(|l| l.to_string()).collect();
        let mut pending_reasoning = String::new();
        let mut pending_reasoning_content = String::new();
        let mut pending_reasoning_encrypted = false;
        // PARITY-15: see `from_claude_code_str`'s identical counter.
        let mut parse_error_lines = 0usize;
        let mut restored_embedded_codex_provenance = false;

        for (record_index, raw_line) in raw_lines.iter().enumerate() {
            let line = raw_line.trim();
            if line.is_empty() {
                continue;
            }
            let v: Value = match serde_json::from_str(line) {
                Ok(v) => v,
                Err(_) => {
                    parse_error_lines += 1;
                    continue;
                }
            };
            let payload = v.get("payload").unwrap_or(&Value::Null);
            if !restored_embedded_codex_provenance
                && v.get("type").and_then(Value::as_str) == Some("session_meta")
                && payload
                    .get(SUPERCODE_NATIVE_RESIDUE_KEY)
                    .map(|extension| restore_native_residue(extension, &mut meta))
                    .or_else(|| {
                        payload
                            .get(SUPERCODE_CODEX_PROVENANCE_KEY)
                            .map(|extension| restore_codex_provenance(extension, &mut meta))
                    })
                    .transpose()?
                    .unwrap_or(false)
            {
                restored_embedded_codex_provenance = true;
            }
            if !restored_embedded_codex_provenance {
                capture_codex_provenance_record(&mut meta, record_index, raw_line, &v);
            }
            // WAVE-2 item 1: every Codex record carries a real top-level
            // `timestamp` (ISO-8601) — threaded onto each `ChatMessage` a
            // line produces via `stamp_new_codex_messages` below, at each
            // arm that pushes messages.
            let line_ts = v.get("timestamp").and_then(Value::as_str);

            match v.get("type").and_then(Value::as_str) {
                Some("session_meta") => {
                    capture_codex_session_meta(payload, &mut meta);
                    if !restored_embedded_codex_provenance {
                        meta.codex_headers.push(v.clone());
                    }
                }
                Some("turn_context") => {
                    if meta.model.is_none() {
                        meta.model = payload
                            .get("model")
                            .and_then(Value::as_str)
                            .map(str::to_string);
                    }
                    if !restored_embedded_codex_provenance {
                        meta.codex_headers.push(v.clone());
                    }
                }
                Some("response_item")
                    if payload.get("type").and_then(Value::as_str) == Some("reasoning") =>
                {
                    // Retain reasoning (P3): summary text if any, the raw
                    // `content` chain-of-thought text if any (N2 — this used
                    // to be dropped despite `Coverage::Retained` claiming the
                    // whole item survived; see `crate::audit`'s doc comment),
                    // plus a flag for the opaque encrypted_content a
                    // same-model continuation can replay. Stashed onto the
                    // next assistant message below.
                    let summary = extract_text_content(payload.get("summary"));
                    if !summary.trim().is_empty() {
                        push_str_field(&mut pending_reasoning, &summary);
                    }
                    // N2: `content` is `null` on the vast majority of real
                    // turns (raw reasoning text is only ever populated for
                    // certain reasoning-transcript configurations) — guard
                    // on non-null BEFORE calling `extract_text_content`,
                    // since `Some(&Value::Null)` would otherwise fall into
                    // its `Some(other) => other.to_string()` arm and
                    // stringify to the literal text `"null"`.
                    if let Some(raw_content) = payload.get("content").filter(|v| !v.is_null()) {
                        let text = extract_text_content(Some(raw_content));
                        if !text.trim().is_empty() {
                            push_str_field(&mut pending_reasoning_content, &text);
                        }
                    }
                    // N1: `serde_json` returns `Some(&Value::Null)` for a
                    // present-but-null `encrypted_content` key — which is
                    // what EVERY real rollout's reasoning item carries
                    // (upstream always serializes the field, never
                    // `skip_serializing_if`, `codex-rs/protocol/src/
                    // models.rs:970-983`). The old `.is_some()` check
                    // false-flagged every single reasoning item as
                    // "encrypted" on real data; only a genuinely non-null
                    // value means the model actually returned an opaque
                    // blob that a same-model continuation could replay.
                    if payload
                        .get("encrypted_content")
                        .is_some_and(|v| !v.is_null())
                    {
                        pending_reasoning_encrypted = true;
                    }
                }
                Some("response_item") => {
                    let before = messages.len();
                    push_codex_item(payload, &mut messages);
                    // Attach any pending reasoning to a newly produced assistant turn.
                    if messages.len() > before
                        && (!pending_reasoning.is_empty()
                            || !pending_reasoning_content.is_empty()
                            || pending_reasoning_encrypted)
                    {
                        let is_assistant = messages
                            .last()
                            .map(|m| m.role == Role::Assistant)
                            .unwrap_or(false);
                        if is_assistant {
                            let last = messages.last_mut().expect("checked above");
                            if !pending_reasoning.is_empty() {
                                last.metadata.insert(
                                    "reasoning".to_string(),
                                    std::mem::take(&mut pending_reasoning),
                                );
                            }
                            if !pending_reasoning_content.is_empty() {
                                last.metadata.insert(
                                    "reasoning_content".to_string(),
                                    std::mem::take(&mut pending_reasoning_content),
                                );
                            }
                            if pending_reasoning_encrypted {
                                last.metadata
                                    .insert("reasoning_encrypted".to_string(), "true".to_string());
                                pending_reasoning_encrypted = false;
                            }
                        } else {
                            // N3: the item that just landed is NOT the
                            // assistant turn the pending reasoning was for
                            // (e.g. an aborted turn's reasoning directly
                            // followed by a user message) — the old code
                            // unconditionally cleared the pending state
                            // here, silently discarding it. Flush it as its
                            // own message instead, inserted just before the
                            // interrupting item so replay order stays
                            // chronological, keeping `Coverage::Retained`
                            // honest for this shape too.
                            let orphan = orphaned_reasoning_message(
                                &mut pending_reasoning,
                                &mut pending_reasoning_content,
                                &mut pending_reasoning_encrypted,
                            );
                            messages.insert(before, orphan);
                        }
                    }
                    stamp_new_codex_messages(&mut messages, before, line_ts);
                    restore_single_grok_message(payload, &mut messages[before..]);
                }
                // A compaction record replaces all prior turns with its
                // summarized `replacement_history` — exactly how Codex itself
                // resumes a compacted session.
                Some("compacted") => {
                    messages.clear();
                    if let Some(Value::Array(history)) = payload.get("replacement_history") {
                        for item in history {
                            push_codex_item(item, &mut messages);
                        }
                    }
                    // `replacement_history` items carry no per-item
                    // timestamp of their own (observed corpora) — the
                    // `compacted` record's own timestamp (when it happened)
                    // is the best-effort real source for every message it
                    // synthesizes, so it stamps the whole rebuilt vec (index
                    // 0, since `clear()` reset it above).
                    stamp_new_codex_messages(&mut messages, 0, line_ts);
                    // IX-6 fix: replaying `replacement_history` through
                    // `push_codex_item` can leave the LAST replayed message
                    // marked `__codex_open_turn` (if it's an assistant
                    // `message`, per the combined-turn merge below). That
                    // marker must not survive past the compaction boundary —
                    // a live `function_call` arriving after this record is a
                    // NEW turn, not a continuation of the compaction
                    // summary's synthetic turn, so it must not merge into it.
                    if let Some(last) = messages.last_mut() {
                        last.metadata.remove("__codex_open_turn");
                    }
                }
                Some("event_msg")
                    if payload.get("type").and_then(Value::as_str) == Some("agent_message") =>
                {
                    let before = messages.len();
                    let text = agent_message_text(payload);
                    if !text.trim().is_empty() && !assistant_texts.contains(text.trim()) {
                        push_assistant(&mut messages, text, Vec::new());
                        if let Some(last) = messages.last_mut() {
                            if let Some(phase) = payload.get("phase").and_then(Value::as_str) {
                                last.metadata.insert("phase".to_string(), phase.to_string());
                            }
                        }
                    }
                    stamp_new_codex_messages(&mut messages, before, line_ts);
                }
                // The user rolled back (undid) the last N turns — replay must
                // drop them so the reloaded conversation matches what the user
                // actually kept.
                Some("event_msg")
                    if payload.get("type").and_then(Value::as_str)
                        == Some("thread_rolled_back") =>
                {
                    let n = payload
                        .get("num_turns")
                        .and_then(Value::as_u64)
                        .unwrap_or(1);
                    for _ in 0..n {
                        remove_last_turn(&mut messages);
                    }
                }
                // The natural-language goal assigned to this thread (sometimes
                // the only place the objective text is recorded).
                Some("event_msg")
                    if payload.get("type").and_then(Value::as_str)
                        == Some("thread_goal_updated") =>
                {
                    let before = messages.len();
                    let goal = payload.get("goal");
                    if let Some(obj) = goal
                        .and_then(|g| g.get("objective"))
                        .and_then(Value::as_str)
                    {
                        if !obj.trim().is_empty() {
                            messages.push(ChatMessage::system(format!("[thread goal] {obj}")));
                            // D4: `goal.objective` alone used to be the ONLY
                            // captured field, but the audit labeled this
                            // `Retained` as if the whole record survived.
                            // `goal.status`/`goal.tokenBudget` (real
                            // `ThreadGoal` wire fields, camelCase) are
                            // captured too so that label is honest — see
                            // `crate::audit::event_msg_coverage`'s doc
                            // comment.
                            if let Some(last) = messages.last_mut() {
                                if let Some(status) =
                                    goal.and_then(|g| g.get("status")).and_then(Value::as_str)
                                {
                                    last.metadata
                                        .insert("goal_status".to_string(), status.to_string());
                                }
                                if let Some(budget) = goal
                                    .and_then(|g| g.get("tokenBudget"))
                                    .and_then(Value::as_i64)
                                {
                                    last.metadata.insert(
                                        "goal_token_budget".to_string(),
                                        budget.to_string(),
                                    );
                                }
                            }
                        }
                    }
                    stamp_new_codex_messages(&mut messages, before, line_ts);
                }
                // Code-review output — unique assistant-generated content with no
                // `message` counterpart.
                Some("event_msg")
                    if payload.get("type").and_then(Value::as_str)
                        == Some("exited_review_mode") =>
                {
                    let before = messages.len();
                    if let Some(review) = payload.get("review_output") {
                        let text = review
                            .get("overall_explanation")
                            .and_then(Value::as_str)
                            .map(str::to_string)
                            .unwrap_or_else(|| review.to_string());
                        push_assistant(&mut messages, format!("[code review] {text}"), Vec::new());
                        // D4: `overall_explanation` alone used to be the ONLY
                        // captured field, but the audit labeled this
                        // `Retained` as if `review_output.findings` survived
                        // too. Capture `findings` verbatim (as JSON, onto
                        // metadata) so that label is honest — this is the
                        // only place review-mode findings (title/body/
                        // confidence_score/priority/code_location) live.
                        if let Some(findings) = review.get("findings") {
                            if findings.as_array().is_some_and(|a| !a.is_empty()) {
                                if let Some(last) = messages.last_mut() {
                                    if let Ok(s) = serde_json::to_string(findings) {
                                        last.metadata.insert("review_findings".to_string(), s);
                                    }
                                }
                            }
                        }
                        // N4: `overall_correctness`/`overall_confidence_score`
                        // are the review's actual verdict — distinct from the
                        // findings list and the explanation prose already
                        // captured above — and were neither captured nor
                        // disclosed as residue while the audit doc stayed
                        // silent about them. Capture both onto the same
                        // message's metadata, same pattern as `findings`.
                        if let Some(last) = messages.last_mut() {
                            if let Some(correctness) =
                                review.get("overall_correctness").and_then(Value::as_str)
                            {
                                last.metadata.insert(
                                    "review_overall_correctness".to_string(),
                                    correctness.to_string(),
                                );
                            }
                            if let Some(score) = review
                                .get("overall_confidence_score")
                                .and_then(Value::as_f64)
                            {
                                last.metadata.insert(
                                    "review_overall_confidence_score".to_string(),
                                    score.to_string(),
                                );
                            }
                        }
                    }
                    stamp_new_codex_messages(&mut messages, before, line_ts);
                }
                _ => {} // other event_msg, token_count, ... — UI events, skip
            }
        }

        // N3/PARITY-11: reasoning still pending at EOF is an aborted-turn
        // shape a real rollout can leave behind (the process was
        // interrupted mid-turn, after the model reasoned but before it
        // replied — end of file, or a rollback/compaction boundary that
        // clears the pending state some other way) — the old code silently
        // dropped it here (nothing ever consumed the pending buffers once
        // the loop ended). Flush it as its own trailing message instead, so
        // `Coverage::Retained` holds for this shape too. Superset of the
        // independently-discovered PARITY-11 fix: also folds in
        // `pending_reasoning_content` (the raw chain-of-thought, distinct
        // from `summary`/`reasoning`) via the shared `orphaned_reasoning_
        // message` helper, which the interrupted-by-a-user-message shape
        // (`orphaned_reasoning_is_flushed_not_discarded`, case A) also
        // relies on — a trailing-EOF-only flush here would miss that case.
        if !pending_reasoning.is_empty()
            || !pending_reasoning_content.is_empty()
            || pending_reasoning_encrypted
        {
            let orphan = orphaned_reasoning_message(
                &mut pending_reasoning,
                &mut pending_reasoning_content,
                &mut pending_reasoning_encrypted,
            );
            messages.push(orphan);
        }

        ensure_tool_results_paired(&mut messages);
        // IX-6: `__codex_open_turn` is an internal bookkeeping marker for the
        // combined-turn merge above — strip it so it never leaks out as
        // visible `ChatMessage` metadata.
        for m in &mut messages {
            m.metadata.remove("__codex_open_turn");
            if m.metadata
                .remove("__grok_remove_synthetic_turn_id")
                .is_some()
            {
                m.metadata.remove("turn_id");
            }
        }
        let imported_message_count = Some(messages.len());
        Ok(Session {
            meta,
            messages,
            subagents: Vec::new(),
            raw,
            raw_trailing_newline,
            imported_message_count,
            // Codex is line-oriented: `raw` is split directly out of the
            // source text (strict-verbatim, IX-1).
            raw_is_verbatim: true,
            parse_error_lines,
            load_residue: Vec::new(),
        })
    }

    /// Parse a Codex rollout as bounded human-visible history rather than as
    /// resumable model context. This deliberately ignores outer `compacted`
    /// replacement semantics: the original `response_item` records remain in
    /// the rollout and are the authoritative UI history.
    pub(super) fn from_codex_display_str(jsonl: &str, message_limit: usize) -> Result<Session> {
        let mut meta = SessionMeta::new(SessionSource::Codex);
        let mut messages: Vec<ChatMessage> = Vec::new();
        let mut preceding_users = Vec::new();
        let mut parse_error_lines = 0usize;
        let mut record_count = 0usize;
        let mut total_message_count = 0usize;
        let retain = message_limit.max(1).saturating_add(64);
        let mut canonical_assistant_texts = HashSet::new();

        for raw_line in non_empty_lines(jsonl) {
            record_count += 1;
            let value: Value = match serde_json::from_str(raw_line) {
                Ok(value) => value,
                Err(_) => {
                    parse_error_lines += 1;
                    continue;
                }
            };
            let payload = value.get("payload").unwrap_or(&Value::Null);
            let line_ts = value.get("timestamp").and_then(Value::as_str);
            match value.get("type").and_then(Value::as_str) {
                Some("session_meta") => capture_codex_session_meta(payload, &mut meta),
                Some("turn_context") if meta.model.is_none() => {
                    meta.model = payload
                        .get("model")
                        .and_then(Value::as_str)
                        .map(str::to_string);
                }
                Some("response_item")
                    if payload.get("type").and_then(Value::as_str) != Some("reasoning") =>
                {
                    let assistant_text = (payload.get("type").and_then(Value::as_str)
                        == Some("message")
                        && payload.get("role").and_then(Value::as_str) == Some("assistant"))
                    .then(|| extract_text_content(payload.get("content")))
                    .filter(|text| !text.trim().is_empty());
                    if let Some(text) = assistant_text.as_deref() {
                        if let Some(index) = messages.iter().rposition(|message| {
                            message.metadata.contains_key("codex_event_message")
                                && message.content.as_deref() == Some(text)
                        }) {
                            messages.remove(index);
                            total_message_count = total_message_count.saturating_sub(1);
                        }
                        canonical_assistant_texts.insert(text.trim().to_string());
                    }
                    let before = messages.len();
                    push_codex_item(payload, &mut messages);
                    total_message_count += messages.len().saturating_sub(before);
                    stamp_new_codex_messages(&mut messages, before, line_ts);
                    restore_single_grok_message(payload, &mut messages[before..]);
                }
                Some("event_msg")
                    if payload.get("type").and_then(Value::as_str) == Some("agent_message") =>
                {
                    let text = agent_message_text(payload);
                    if !text.trim().is_empty() && !canonical_assistant_texts.contains(text.trim()) {
                        let before = messages.len();
                        push_assistant(&mut messages, text, Vec::new());
                        total_message_count += 1;
                        if let Some(last) = messages.last_mut() {
                            last.metadata
                                .insert("codex_event_message".to_string(), "true".to_string());
                            if let Some(phase) = payload.get("phase").and_then(Value::as_str) {
                                last.metadata.insert("phase".to_string(), phase.to_string());
                            }
                        }
                        stamp_new_codex_messages(&mut messages, before, line_ts);
                    }
                }
                // `compacted` changes continuation context, not what was
                // already visible in scrollback. Other event records are UI
                // lifecycle noise or duplicate canonical response items.
                _ => {}
            }
            if messages.len() > retain {
                let remove = messages.len() - retain;
                for message in messages.drain(..remove) {
                    if message.role == Role::User {
                        preceding_users.push(message);
                        if preceding_users.len() > 2 {
                            preceding_users.remove(0);
                        }
                    }
                }
            }
        }

        for message in &mut messages {
            message.metadata.remove("__codex_open_turn");
            message.metadata.remove("codex_event_message");
            if message
                .metadata
                .remove("__grok_remove_synthetic_turn_id")
                .is_some()
            {
                message.metadata.remove("turn_id");
            }
        }
        truncate_messages_with_anchor(&mut messages, message_limit, preceding_users);
        let imported_message_count = Some(total_message_count);
        Ok(Session {
            meta,
            messages,
            subagents: Vec::new(),
            // Preserve the cheap count without retaining hundreds of
            // megabytes of source lines in a display-only value.
            raw: vec![String::new(); record_count],
            raw_trailing_newline: jsonl.ends_with('\n'),
            imported_message_count,
            raw_is_verbatim: false,
            parse_error_lines,
            load_residue: vec![
                "display history is a bounded native-record projection, not resumable model context"
                    .to_string(),
            ],
        })
    }
}

// ---- Codex ----------------------------------------------------------------

pub(super) const SUPERCODE_CODEX_PROVENANCE_KEY: &str = "_supercode_codex_provenance";

fn codex_provenance_kind(record: &Value) -> Option<&str> {
    match record.get("type").and_then(Value::as_str) {
        Some("session_meta") => Some("session_meta"),
        Some("turn_context") => Some("turn_context"),
        Some("compacted") => Some("compacted"),
        Some("event_msg") => match record
            .get("payload")
            .and_then(|payload| payload.get("type"))
            .and_then(Value::as_str)
        {
            Some("thread_rolled_back") => Some("event_msg/thread_rolled_back"),
            Some("thread_goal_updated") => Some("event_msg/thread_goal_updated"),
            Some("entered_review_mode") => Some("event_msg/entered_review_mode"),
            Some("exited_review_mode") => Some("event_msg/exited_review_mode"),
            _ => None,
        },
        _ => None,
    }
}

fn capture_codex_provenance_record(
    meta: &mut SessionMeta,
    record_index: usize,
    raw_line: &str,
    record: &Value,
) {
    let Some(kind) = codex_provenance_kind(record) else {
        return;
    };
    meta.codex_provenance.push(serde_json::json!({
        "record_index": record_index,
        "kind": kind,
        "raw": raw_line,
    }));
}

fn codex_provenance_envelope(meta: &SessionMeta) -> Option<Value> {
    (!meta.codex_provenance.is_empty()).then(|| {
        serde_json::json!({
            "version": 1,
            "records": &meta.codex_provenance,
        })
    })
}

pub(super) fn restore_codex_provenance(extension: &Value, meta: &mut SessionMeta) -> Result<bool> {
    if extension.get("version").and_then(Value::as_u64) != Some(1) {
        return Err(Error::InvalidSession(
            "invalid portable Codex provenance: expected version 1".to_string(),
        ));
    }
    let Some(records) = extension.get("records").and_then(Value::as_array) else {
        return Err(Error::InvalidSession(
            "invalid portable Codex provenance: `records` must be an array".to_string(),
        ));
    };
    if records.is_empty() {
        return Err(Error::InvalidSession(
            "invalid portable Codex provenance: `records` must not be empty".to_string(),
        ));
    }
    let mut restored = Vec::with_capacity(records.len());
    for entry in records {
        let Some(_record_index) = entry.get("record_index").and_then(Value::as_u64) else {
            return Err(Error::InvalidSession(
                "invalid portable Codex provenance: record_index must be an integer".to_string(),
            ));
        };
        let Some(kind) = entry.get("kind").and_then(Value::as_str) else {
            return Err(Error::InvalidSession(
                "invalid portable Codex provenance: kind must be a string".to_string(),
            ));
        };
        let Some(raw) = entry.get("raw").and_then(Value::as_str) else {
            return Err(Error::InvalidSession(
                "invalid portable Codex provenance: raw must be a string".to_string(),
            ));
        };
        let Ok(record) = serde_json::from_str::<Value>(raw) else {
            return Err(Error::InvalidSession(
                "invalid portable Codex provenance: raw is not valid JSON".to_string(),
            ));
        };
        if codex_provenance_kind(&record) != Some(kind) {
            return Err(Error::InvalidSession(format!(
                "invalid portable Codex provenance: kind `{kind}` does not match raw record"
            )));
        }
        restored.push(entry.clone());
    }
    meta.codex_provenance = restored;
    meta.codex_headers.clear();
    for entry in &meta.codex_provenance {
        let Some(raw) = entry.get("raw").and_then(Value::as_str) else {
            continue;
        };
        let Ok(record) = serde_json::from_str::<Value>(raw) else {
            continue;
        };
        if matches!(
            record.get("type").and_then(Value::as_str),
            Some("session_meta") | Some("turn_context")
        ) {
            meta.codex_headers.push(record);
        }
    }
    Ok(true)
}

pub(super) fn restore_codex_provenance_from_top_level(
    record: &Value,
    meta: &mut SessionMeta,
) -> Result<bool> {
    if let Some(extension) = record.get(SUPERCODE_NATIVE_RESIDUE_KEY) {
        return restore_native_residue(extension, meta);
    }
    if let Some(extension) = record.get(SUPERCODE_CODEX_PROVENANCE_KEY) {
        return restore_codex_provenance(extension, meta);
    }
    if let Some(summary) = record.get(SUPERCODE_NATIVE_RESIDUE_SUMMARY_KEY) {
        // Tombstone without its records: the residue was deliberately
        // deleted. The transcript stays fully usable; the loss is REPORTED,
        // never silent (PARITY-23 dev/05).
        let kinds = summary
            .get("kinds")
            .and_then(Value::as_array)
            .map(|kinds| {
                kinds
                    .iter()
                    .filter_map(Value::as_str)
                    .collect::<Vec<_>>()
                    .join(", ")
            })
            .unwrap_or_default();
        let count = summary.get("records").and_then(Value::as_u64).unwrap_or(0);
        let source = summary
            .get("source")
            .and_then(Value::as_str)
            .unwrap_or("unknown");
        meta.lineage.insert(
            "residue_loss".to_string(),
            format!(
                "portable {source} residue deleted: {count} record(s) of kind(s) [{kinds}] \
                 can no longer be restored"
            ),
        );
        return Ok(false);
    }
    Ok(false)
}

fn inject_codex_provenance(out: &mut String, extension: Value) {
    let Some(line_end) = out.find('\n') else {
        return;
    };
    let Ok(mut record) = serde_json::from_str::<Value>(&out[..line_end]) else {
        return;
    };
    if record.get("type").and_then(Value::as_str) != Some("session_meta") {
        return;
    }
    let Some(payload) = record.get_mut("payload").and_then(Value::as_object_mut) else {
        return;
    };
    payload.insert(
        SUPERCODE_NATIVE_RESIDUE_SUMMARY_KEY.to_string(),
        native_residue_summary(&extension),
    );
    payload.insert(SUPERCODE_NATIVE_RESIDUE_KEY.to_string(), extension);
    out.replace_range(..line_end, &record.to_string());
}

/// The text of a Codex `agent_message` event. `message` is usually a string but
/// can be a structured object (e.g. review output) — fall back to its JSON.
fn agent_message_text(payload: &Value) -> String {
    match payload.get("message") {
        Some(Value::String(s)) => s.clone(),
        Some(other) => extract_text_content(Some(other)),
        None => String::new(),
    }
}

/// Trimmed texts of all assistant messages present as `response_item` — the
/// dedup set for recovering collab-only `agent_message` narration.
fn collect_codex_assistant_texts(jsonl: &str) -> std::collections::HashSet<String> {
    let mut set = std::collections::HashSet::new();
    for line in non_empty_lines(jsonl) {
        let Ok(v) = serde_json::from_str::<Value>(line) else {
            continue;
        };
        if v.get("type").and_then(Value::as_str) != Some("response_item") {
            continue;
        }
        let payload = v.get("payload").unwrap_or(&Value::Null);
        if payload.get("type").and_then(Value::as_str) == Some("message")
            && payload.get("role").and_then(Value::as_str) == Some("assistant")
        {
            let text = extract_text_content(payload.get("content"));
            if !text.trim().is_empty() {
                set.insert(text.trim().to_string());
            }
        }
    }
    set
}

fn capture_codex_session_meta(payload: &Value, meta: &mut SessionMeta) {
    if meta.session_id.is_none() {
        if let Some(id) = payload.get("id").and_then(Value::as_str) {
            meta.session_id = Some(id.to_string());
        }
    }
    if meta.cwd.is_none() {
        if let Some(cwd) = payload.get("cwd").and_then(Value::as_str) {
            meta.cwd = Some(PathBuf::from(cwd));
        }
    }
    if meta.system_prompt.is_none() {
        // `base_instructions` may be a string or `{ "text": "..." }`.
        let bi = payload.get("base_instructions");
        let text = match bi {
            Some(Value::String(s)) => Some(s.clone()),
            Some(Value::Object(_)) => bi
                .and_then(|b| b.get("text"))
                .and_then(Value::as_str)
                .map(str::to_string),
            _ => None,
        };
        meta.system_prompt = text;
    }
    if meta.model.is_none() {
        if let Some(m) = payload.get("model").and_then(Value::as_str) {
            meta.model = Some(m.to_string());
        }
    }
    // Cross-file lineage keys for multi-agent / forked sessions.
    let mut put = |key: &str, v: Option<&Value>| {
        if let Some(s) = v.and_then(Value::as_str) {
            meta.lineage.insert(key.to_string(), s.to_string());
        }
    };
    put("parent_thread_id", payload.get("parent_thread_id"));
    put("forked_from_id", payload.get("forked_from_id"));
    put("thread_source", payload.get("thread_source"));
    // PARITY-10 dev/03: the other half of `write_synthesized_codex_header`'s
    // passthrough — restores a captured Claude `fork-context-ref` so a
    // Claude -> Codex -> Claude round trip reconstructs the original record
    // (`to_claude_code_jsonl` re-emits whatever lands in this lineage key).
    if !meta.lineage.contains_key("claude_fork_context_ref_raw") {
        if let Some(v) = payload.get("claude_fork_context_ref") {
            meta.lineage
                .insert("claude_fork_context_ref_raw".to_string(), v.to_string());
        }
    }
    if let Some(spawn) = payload
        .get("source")
        .and_then(|s| s.get("subagent"))
        .and_then(|s| s.get("thread_spawn"))
    {
        // parent_thread_id can also live here (preferred when both present).
        if let Some(p) = spawn.get("parent_thread_id").and_then(Value::as_str) {
            meta.lineage
                .insert("parent_thread_id".to_string(), p.to_string());
        }
        for k in ["agent_role", "agent_nickname"] {
            if let Some(s) = spawn.get(k).and_then(Value::as_str) {
                meta.lineage.insert(k.to_string(), s.to_string());
            }
        }
        if let Some(d) = spawn.get("depth").and_then(Value::as_i64) {
            meta.lineage.insert("depth".to_string(), d.to_string());
        }
    }
}

/// Codex per-turn grouping id, stored under `payload.metadata.turn_id`.
pub(super) fn codex_turn_id(payload: &Value) -> Option<&str> {
    payload
        .get("metadata")
        .and_then(|m| m.get("turn_id"))
        .and_then(Value::as_str)
}

/// N2 (spliced-export hardening): every Codex group id already present in
/// `raw_prefix` — the verbatim RAW lines [`Session::to_codex_jsonl_spliced`]
/// replays ahead of the appended tail it synthesizes via
/// `Session::write_codex_records`. This is the GROUND TRUTH of what
/// physically lands in the exported `out` string for the prefix: each line
/// is parsed as a Codex envelope and its own `payload.metadata.turn_id` (the
/// exact field `codex_turn_id` reads, whether it's a real native `turn_id`
/// or one of OUR OWN fabricated `sc-grp-N`/`<real>~dupN` ids from a prior
/// export) is extracted directly — no re-derivation from `self.messages`
/// needed (that would have to reconstruct which ids the ORIGINAL export
/// happened to assign, which this sidesteps entirely by reading them back
/// out of the bytes themselves). A line that fails to parse, isn't a
/// `response_item`, or carries no `turn_id` contributes nothing — headers
/// and non-message/call records (e.g. `session_meta`, `function_call_output`)
/// never carry this field to begin with.
fn collect_codex_group_ids_from_raw(raw_prefix: &[String]) -> HashSet<String> {
    let mut ids = HashSet::new();
    for line in raw_prefix {
        if let Ok(v) = serde_json::from_str::<Value>(line) {
            if let Some(payload) = v.get("payload") {
                if let Some(tid) = codex_turn_id(payload) {
                    ids.insert(tid.to_string());
                }
            }
        }
    }
    ids
}

/// Stamp every `ChatMessage` appended to `messages` since index `from` with
/// `ts` (a Codex record's own top-level `timestamp`, ISO-8601) as the
/// canonical `metadata["timestamp"]` (WAVE-2 item 1) — the same
/// `entry(...).or_insert_with` discipline the pi/Claude loaders use, so a
/// message that already carries a more specific timestamp of its own is
/// never overwritten (none currently do on the Codex side, but this keeps
/// every loader consistent). A no-op when `ts` is `None` (a line with no
/// `timestamp` field) or `from >= messages.len()` (nothing new was pushed).
fn stamp_new_codex_messages(messages: &mut [ChatMessage], from: usize, ts: Option<&str>) {
    let Some(ts) = ts else { return };
    let Some(slice) = messages.get_mut(from..) else {
        return;
    };
    for m in slice {
        m.metadata
            .entry("timestamp".to_string())
            .or_insert_with(|| ts.to_string());
    }
}

fn push_codex_item(payload: &Value, out: &mut Vec<ChatMessage>) {
    match payload.get("type").and_then(Value::as_str) {
        Some("message") => {
            let role = match payload.get("role").and_then(Value::as_str) {
                Some("user") => Role::User,
                Some("assistant") => Role::Assistant,
                // "developer" and "system" both carry operator instructions.
                _ => Role::System,
            };
            let content = payload.get("content");
            let text = extract_text_content(content);
            // IX-5: `input_image` blocks alongside/instead of text — see
            // `codex_extract_images`. A text-only message (no image blocks)
            // takes the historical `content: Some(text)` shape unchanged.
            let images = codex_extract_images(content);
            let is_empty_assistant =
                role == Role::Assistant && text.trim().is_empty() && images.is_empty();
            if !text.trim().is_empty() || !images.is_empty() || is_empty_assistant {
                let content_parts = if images.is_empty() {
                    None
                } else {
                    let mut parts = Vec::new();
                    if !text.trim().is_empty() {
                        parts.push(serde_json::json!({"type": "text", "text": text.clone()}));
                    }
                    parts.extend(images);
                    Some(parts)
                };
                let mut msg = ChatMessage {
                    role,
                    content: if content_parts.is_some() || text.is_empty() {
                        None
                    } else {
                        Some(text)
                    },
                    content_parts,
                    tool_calls: None,
                    tool_call_id: None,
                    name: None,
                    metadata: Default::default(),
                };
                // Preserve the assistant `phase` (commentary vs final_answer) so
                // a reloaded transcript can distinguish narration from the answer.
                if role == Role::Assistant {
                    if let Some(phase) = payload.get("phase").and_then(Value::as_str) {
                        msg.metadata.insert("phase".to_string(), phase.to_string());
                    }
                    // IX-6: mark this as an open, mergeable combined-turn
                    // candidate — a `function_call` response_item found
                    // immediately after (still `out.last()` when reached,
                    // i.e. no other item intervened) merges into this SAME
                    // `ChatMessage` instead of splitting into a second one,
                    // matching how Claude's parser keeps a text+tool_use
                    // turn together. Stripped again before the loaded
                    // `Session` is returned (`from_codex_str`), so it never
                    // leaks as visible metadata.
                    msg.metadata
                        .insert("__codex_open_turn".to_string(), "true".to_string());
                }
                // The per-turn grouping key (Codex batches items by turn_id).
                if let Some(tid) = codex_turn_id(payload) {
                    msg.metadata.insert("turn_id".to_string(), tid.to_string());
                }
                // PARITY-6 dev/02: restore the original Claude
                // `systemSubtype` for a `developer`/`system` message that
                // was itself synthesized FROM a real Claude system record
                // (`write_codex_records`'s `Role::System` arm stamps
                // `claude_system_subtype`) — the exact inverse, so
                // `write_claude_code_records`'s `Role::System` arm can
                // re-materialize the real Claude `type: "system"` record
                // faithfully on a Codex -> Claude Code hop instead of
                // guessing a fallback subtype.
                if role == Role::System {
                    if let Some(subtype) = payload
                        .get("metadata")
                        .and_then(|m| m.get("claude_system_subtype"))
                        .and_then(Value::as_str)
                    {
                        msg.metadata
                            .insert("systemSubtype".to_string(), subtype.to_string());
                    }
                }
                if is_empty_assistant {
                    msg.metadata
                        .insert("empty_assistant_record".to_string(), "true".to_string());
                }
                out.push(msg);
            }
        }
        Some("function_call") => {
            let id = payload
                .get("call_id")
                .and_then(Value::as_str)
                .unwrap_or_default();
            let raw_name = payload
                .get("name")
                .and_then(Value::as_str)
                .unwrap_or_default();
            // Preserve the MCP `namespace` by qualifying the tool name
            // (`<namespace>__<name>`, matching the mcp__server__tool convention),
            // so the tool identity isn't ambiguous on round-trip.
            let qualified;
            let name = match payload.get("namespace").and_then(Value::as_str) {
                Some(ns) if !ns.is_empty() && !raw_name.starts_with(ns) => {
                    qualified = format!("{ns}__{raw_name}");
                    qualified.as_str()
                }
                _ => raw_name,
            };
            let args = payload
                .get("arguments")
                .map(value_to_arg_string)
                .unwrap_or_else(|| "{}".to_string());
            let call = function_call(id, name, args);
            // IX-6: a `function_call` immediately after an assistant `message`
            // in the SAME turn (still `out.last()`, marked `__codex_open_turn`
            // by the "message" arm above, and not yet closed by anything else)
            // merges into that ONE `ChatMessage` — text→`content`,
            // call→`tool_calls` — instead of splitting into a second message.
            // A bare `function_call` with no such preceding turn (the marker
            // absent, or `out.last()` not an assistant message) is unaffected:
            // it still gets its own synthesized message, exactly as before.
            //
            // Belt-and-suspenders (PARITY-6/7 tightened): if this
            // `function_call` response_item itself carries a `turn_id` (rare
            // in observed real-native-Codex corpora — Codex usually only
            // stamps it on `message` payloads — but ALWAYS present on OUR
            // OWN synthesized export whenever a `ChatMessage`'s own tool
            // calls need merge disambiguation, see `write_codex_records`),
            // it must match the marked assistant message's recorded
            // `turn_id` EXACTLY — including "the marked message has none at
            // all" counting as a mismatch. That's exactly the shape of two
            // genuinely separate, adjacent `ChatMessage`s (an unrelated
            // text-only turn immediately followed by a different,
            // tool-call-only turn): the tool-only turn's own `function_call`s
            // carry a synthetic id while the unrelated preceding text
            // message carries none, so this correctly refuses the merge
            // instead of falling through to a permissive default. Only when
            // this `function_call` carries NO `turn_id` at all (the ordinary
            // real-native-Codex shape) does this fall back to the original
            // permissive "adjacency + open marker is enough" rule —
            // unchanged from before for the vast majority of real Codex
            // data. The truncation/clear strip above is what actually closes
            // the marker across rollback/compaction boundaries; this is only
            // an extra guard for the case where a stale-but-unstripped
            // marker and a turn_id mismatch coincide.
            let can_merge = out.last().is_some_and(|last| {
                last.role == Role::Assistant
                    && last.metadata.contains_key("__codex_open_turn")
                    && match codex_turn_id(payload) {
                        Some(fc_tid) => {
                            last.metadata.get("turn_id").map(String::as_str) == Some(fc_tid)
                        }
                        None => true,
                    }
            });
            if can_merge {
                out.last_mut()
                    .expect("can_merge implies out.last() is Some")
                    .tool_calls
                    .get_or_insert_with(Vec::new)
                    .push(call);
            } else {
                push_assistant(out, String::new(), vec![call]);
                // PARITY-6/7: a BARE tool-call turn (no preceding `message`
                // in this turn, so nothing set `__codex_open_turn` above) can
                // still be the FIRST of several tool calls that all belong to
                // the SAME original `ChatMessage` (`write_codex_records`
                // stamps every one of a message's own tool calls with the
                // identical synthetic `turn_id`). Re-open THIS freshly
                // created message — but ONLY when a real `turn_id` is
                // present — so the NEXT `function_call` in the same group
                // merges into it instead of becoming its own message too.
                // Gated on `codex_turn_id(payload).is_some()` (not the bare
                // default `true` the belt-and-suspenders check above uses)
                // so real native Codex data — which almost never carries
                // this field on `function_call` payloads (see the comment
                // above) — keeps its existing "every bare tool call is its
                // own turn" behavior exactly as before.
                if let Some(tid) = codex_turn_id(payload) {
                    if let Some(last) = out.last_mut() {
                        last.metadata
                            .insert("__codex_open_turn".to_string(), "true".to_string());
                        last.metadata.insert("turn_id".to_string(), tid.to_string());
                    }
                }
            }
        }
        Some("function_call_output") => {
            let id = payload
                .get("call_id")
                .and_then(Value::as_str)
                .unwrap_or_default();
            let result = match payload.get("output") {
                Some(Value::String(s)) => s.clone(),
                Some(v) => extract_text_content(Some(v)),
                None => String::new(),
            };
            let mut message = tool_message(id, result);
            // TR-13: Codex v1 exposes no structured success/error field on
            // this record. Free-text output is not a safe classifier, so the
            // reduction engine must treat the outcome as explicitly unknown
            // and fail closed on both success-only and error-only pruning.
            crate::mark_tool_outcome_unknown(&mut message);
            out.push(message);
        }
        // Custom / MCP tool calls are shaped like function calls but carry their
        // arguments under `input` (a JSON-encoded string). Normalize them the
        // same way so MCP-using sessions don't lose those turns.
        Some("custom_tool_call") => {
            let id = payload
                .get("call_id")
                .and_then(Value::as_str)
                .unwrap_or_default();
            let name = payload
                .get("name")
                .and_then(Value::as_str)
                .unwrap_or_default();
            // Unlike `function_call.arguments`, Codex custom tools accept a
            // free-form `input` string (apply_patch is the common case).
            // Canonical `FunctionCall::arguments` must remain valid JSON, so
            // retain the input's JSON type instead of treating a free-form
            // string as if it were already a JSON document. This lets every
            // target harness carry the value rather than silently replacing
            // it with `{}` when `parsed_arguments()` fails.
            let args = payload
                .get("input")
                .map(Value::to_string)
                .unwrap_or_else(|| "{}".to_string());
            push_assistant(out, String::new(), vec![function_call(id, name, args)]);
            if let Some(message) = out.last_mut() {
                message.metadata.insert(
                    "codex_custom_tool_call_ids".to_string(),
                    serde_json::json!([id]).to_string(),
                );
            }
        }
        Some("custom_tool_call_output") => {
            let id = payload
                .get("call_id")
                .and_then(Value::as_str)
                .unwrap_or_default();
            let result = match payload.get("output") {
                Some(Value::String(s)) => s.clone(),
                Some(v) => extract_text_content(Some(v)),
                None => String::new(),
            };
            let mut message = tool_message(id, result);
            crate::mark_tool_outcome_unknown(&mut message);
            out.push(message);
        }
        // Tool-search is a clean call/output pair keyed by call_id.
        //
        // D1: this arm used to ALWAYS start a brand-new `ChatMessage`,
        // ignoring the `turn_id` merge stamps `write_codex_records` puts on
        // its own synthesized `tool_search_call` records (see the PARITY-6/7
        // comment there and on `codex_turn_id`/the `function_call` arm
        // above). That left the same bug-class the turn_id work fixed for
        // `function_call` half-done here: a single Claude assistant message
        // containing text + a `tool_search` block reloaded as 2 messages
        // (1 -> 2 inflation), and a message with 2 `tool_search` blocks
        // reloaded as 3. Mirror the `function_call` arm's merge check
        // exactly so a `tool_search_call` immediately following an open
        // assistant turn (or another tool call sharing the same `turn_id`)
        // merges into that SAME `ChatMessage` instead of splitting.
        Some("tool_search_call") => {
            let id = payload
                .get("call_id")
                .and_then(Value::as_str)
                .unwrap_or_default();
            let args = payload
                .get("arguments")
                .map(value_to_arg_string)
                .unwrap_or_else(|| "{}".to_string());
            let call = function_call(id, "tool_search", args);
            let can_merge = out.last().is_some_and(|last| {
                last.role == Role::Assistant
                    && last.metadata.contains_key("__codex_open_turn")
                    && match codex_turn_id(payload) {
                        Some(fc_tid) => {
                            last.metadata.get("turn_id").map(String::as_str) == Some(fc_tid)
                        }
                        None => true,
                    }
            });
            if can_merge {
                out.last_mut()
                    .expect("can_merge implies out.last() is Some")
                    .tool_calls
                    .get_or_insert_with(Vec::new)
                    .push(call);
            } else {
                push_assistant(out, String::new(), vec![call]);
                // Re-open the freshly created message so a FOLLOWING
                // `function_call`/`tool_search_call` sharing this same
                // `turn_id` merges into it too — matching the bare
                // `function_call` case's own re-open logic above.
                if let Some(tid) = codex_turn_id(payload) {
                    if let Some(last) = out.last_mut() {
                        last.metadata
                            .insert("__codex_open_turn".to_string(), "true".to_string());
                        last.metadata.insert("turn_id".to_string(), tid.to_string());
                    }
                }
            }
        }
        Some("tool_search_output") => {
            let id = payload
                .get("call_id")
                .and_then(Value::as_str)
                .unwrap_or_default();
            let result = payload
                .get("tools")
                .map(value_to_arg_string)
                .unwrap_or_default();
            out.push(tool_message(id, result));
        }
        // Web-search / image-generation response_items carry no paired output
        // here (results live in event_msg), so emit an assistant marker rather
        // than a dangling unanswered tool call.
        Some("web_search_call") => {
            push_assistant(out, "[web_search]".to_string(), Vec::new());
        }
        Some("image_generation_call") => {
            let prompt = payload
                .get("revised_prompt")
                .and_then(Value::as_str)
                .unwrap_or("");
            push_assistant(
                out,
                format!("[image_generation] {prompt}").trim().to_string(),
                Vec::new(),
            );
        }
        // "reasoning" and anything else — dropped.
        _ => {}
    }
}

impl Session {
    /// Synthesize a Codex rollout.
    pub(super) fn to_codex_jsonl(&self) -> String {
        let mut out = String::new();

        if self.meta.codex_headers.is_empty() {
            self.write_synthesized_codex_header(&mut out);
        } else {
            // Replay the exact header records the original tool wrote — Codex's
            // reader validates the header shape strictly — overriding only the
            // session id when the caller changed it.
            for header in &self.meta.codex_headers {
                let mut header = header.clone();
                if header.get("type").and_then(Value::as_str) == Some("session_meta") {
                    if let Some(id) = &self.meta.session_id {
                        if let Some(payload) = header.get_mut("payload") {
                            payload["id"] = Value::String(id.clone());
                        }
                    }
                }
                push_jsonl(&mut out, &header);
            }
        }

        // Full synthesis: `out` at this point is only the header, so there
        // are no group ids yet in play to seed against (see
        // `write_codex_records`'s doc comment).
        self.write_codex_records(&mut out, &self.messages, &std::collections::HashSet::new());
        if let Some(extension) = native_residue_envelope(&self.meta) {
            inject_codex_provenance(&mut out, extension);
        }
        out
    }

    /// Synthesize Codex `response_item` records for `messages` (a full
    /// session or an appended tail — A12's [`Self::to_jsonl_spliced`] reuses
    /// this for just the latter). Factored out of [`Self::to_codex_jsonl`] so
    /// the record shape is defined once; `tool_search_call_ids` pairing is
    /// scoped to this call's `messages`, matching the header-replay
    /// contract that only appended records need synthesizing.
    ///
    /// `seed_used_ids` primes the N2 collision guard below with every group
    /// id that will ALREADY be present in `out` before this call ever runs —
    /// [`Self::to_codex_jsonl`] (full synthesis, `out` starts as just the
    /// header) passes an empty set, since every group id in that case is
    /// assigned by this very loop. [`Self::to_codex_jsonl_spliced`] (the A12
    /// splice) passes the ids already used by the verbatim RAW prefix it
    /// replayed into `out` just before calling this for the appended tail —
    /// without that seed, the tail's own `used_group_ids`/`next_group_id`
    /// start blind to the prefix and can fabricate/reuse a group id that
    /// COLLIDES with one still "open" at the end of the prefix, letting
    /// reimport's merge check (`can_merge`, `push_codex_item`) wrongly splice
    /// an unrelated appended message into a historical one — the same
    /// bug-class N2 closed for full synthesis, reopened here because the
    /// spliced tail's tracking set used to always start empty regardless of
    /// what the replayed prefix already contained.
    fn write_codex_records(
        &self,
        out: &mut String,
        messages: &[ChatMessage],
        seed_used_ids: &std::collections::HashSet<String>,
    ) {
        // Call ids of assistant `tool_search` calls (B6 agent intrinsic), so
        // the matching tool result below can be emitted as the paired
        // `tool_search_output` record rather than a generic
        // `function_call_output` — the exact inverse of the importer's
        // `tool_search_call`/`tool_search_output` normalization
        // (`push_codex_item`, above).
        let mut tool_search_call_ids: std::collections::HashSet<String> = Default::default();
        // PARITY-6/7: two ADJACENT but genuinely SEPARATE Claude assistant
        // records (e.g. a text-only narration turn immediately followed by a
        // bare tool-call turn, no user turn between — a real, common Claude
        // Code shape) each become their own Codex `message`/`function_call`
        // response_item(s) here. Codex's own reader (`push_codex_item`, IX-6)
        // opportunistically RE-MERGES an assistant `message` immediately
        // followed by a `function_call` back into ONE `ChatMessage`, to match
        // how a genuinely single Claude turn (text+tool_use in the SAME
        // record) round-trips — but with no distinguishing signal, it can't
        // tell that case apart from two originally-separate records that
        // just happen to be adjacent, so it wrongly recombines them too,
        // silently shrinking the message count on every Claude -> Codex ->
        // (inspect) hop. Fix: stamp a synthetic `metadata.turn_id` — unique
        // per ORIGINAL `ChatMessage` — onto the `message` record AND every
        // `function_call`/`tool_search_call` record THAT SAME `ChatMessage`
        // itself emits. `push_codex_item`'s merge already treats a turn_id
        // mismatch as "different turn, do not merge" (the pre-existing
        // belt-and-suspenders check); real native Codex data almost never
        // carries this field (per that check's own comment), so this is a
        // no-op there and only sharpens fidelity for OUR OWN synthesized
        // export.
        let mut next_group_id: u64 = 0;
        // N2 (Fable-5 review, turn_id-collision hardening): every group id
        // this export has already assigned — whether REUSED from a real
        // `turn_id` or FABRICATED as `sc-grp-N` — so a later assistant
        // `ChatMessage` never emits one that's already in use. Two concrete
        // mis-merge scenarios motivate this:
        //
        // (a) Claude->Codex export fabricates `sc-grp-0` for msg A (msg A's
        //     own text+tool_use); reload makes A carry REAL turn_id
        //     `sc-grp-0`. A bare tool-call msg B (metadata has no turn_id of
        //     its own) is then appended. Re-export: A reuses its real
        //     `sc-grp-0`, but B independently fabricates a FRESH id starting
        //     from `next_group_id == 0` again (nothing bumped it when A's id
        //     was reused rather than fabricated) — also `sc-grp-0`.
        //     Collision. If A's call has no output (interrupted session),
        //     reimport sees message(sc-grp-0)+call(sc-grp-0)+call(sc-grp-0)
        //     adjacent with nothing to break the run and merges all three
        //     into ONE message (2 -> 1).
        // (b) Native Codex: `message(turn-7)` opens the merge marker, a
        //     truncation/clear event strips `__codex_open_turn` (closing the
        //     turn without changing the id), then `function_call(turn-7)`
        //     loads as a SECOND, separate `ChatMessage` that still carries
        //     the SAME real `turn_id` (the reopen step in `push_codex_item`
        //     restamps it). Full-synthesis export naively reuses `turn-7`
        //     verbatim for BOTH messages (they're two different loop
        //     iterations, each independently reusing its own `real_turn_id`)
        //     and emits them adjacent — reimport's merge check can't tell
        //     this apart from a single message's own multi-call turn and
        //     recombines them (2 -> 1).
        //
        // Fix: the fabricated-id counter is advanced (skipped) past any id
        // already in `used_group_ids`, AND a real id that's already been
        // used gets disambiguated (`<real>~dupN`) instead of reused verbatim
        // — never letting two DIFFERENT `ChatMessage`s in this export share
        // one group id, since `push_codex_item`'s merge check treats a
        // shared id as "same turn, merge". A single `ChatMessage`'s own
        // message record + its own tool call records still share ONE group
        // id (computed once per loop iteration below, before insertion), so
        // the D1 tool_search merge and ordinary same-turn multi-call
        // grouping are unaffected — this only stops REUSE across iterations.
        //
        // Seeded from `seed_used_ids` (see this fn's doc comment) so the
        // spliced-export tail is likewise blind-proof against the prefix it
        // doesn't itself write.
        let mut used_group_ids: std::collections::HashSet<String> = seed_used_ids.clone();

        for msg in messages {
            if is_replay_excluded(msg) {
                continue;
            }
            // D3 (Fable-5 review): a message loaded FROM real native Codex
            // carries its OWN real `turn_id` in `msg.metadata["turn_id"]`
            // (`push_codex_item`'s "message" arm stamps it whenever the
            // source record itself has one). The group-id logic below used
            // to ALWAYS fabricate a fresh `"sc-grp-N"` value regardless,
            // silently overwriting/discarding that real id on any
            // native-Codex -> load -> export-Codex hop. Reuse it verbatim
            // when present; only fabricate a synthetic id as a fallback for
            // our own merge-disambiguation need (PARITY-6/7) when the
            // message has no real one of its own.
            let real_turn_id = msg.metadata.get("turn_id").map(String::as_str);
            match msg.role {
                Role::System => {
                    // PARITY-6 dev/02: carry the original Claude
                    // `systemSubtype` (`push_claude_system`'s `.with_meta`)
                    // through as `metadata.claude_system_subtype`, so
                    // `push_codex_item`'s reverse load can restore it and
                    // `write_claude_code_records`'s `Role::System` arm can
                    // re-materialize the EXACT original subtype rather than
                    // guessing on a Codex -> Claude hop.
                    let subtype_meta = msg
                        .metadata
                        .get("systemSubtype")
                        .map(|s| ("claude_system_subtype", s.as_str()));
                    self.push_codex_message(
                        out,
                        "developer",
                        "input_text",
                        msg,
                        real_turn_id,
                        subtype_meta,
                    )
                }
                Role::User => {
                    self.push_codex_message(out, "user", "input_text", msg, real_turn_id, None)
                }
                Role::Assistant => {
                    // Emit the message record whenever there is text OR
                    // content_parts (IX-6 follow-up): an image-only assistant
                    // message has `content: None, content_parts:
                    // Some([image])` (the loader's `codex_extract_images` is
                    // role-general, so this shape can occur on the assistant
                    // side too) — gating on `msg.content` alone silently
                    // dropped the whole message, image included. A
                    // text-only message (content_parts: None) keeps taking
                    // the historical byte-identical path via
                    // `codex_message_content_blocks`'s `None` arm. A real
                    // empty native assistant record carries the
                    // loader's explicit marker and must also be emitted.
                    // Reasoning-only cross-provider turns deliberately lack
                    // that marker and keep the documented Codex residue.
                    let has_text = msg.content.as_deref().is_some_and(|t| !t.is_empty());
                    let has_message_record = has_text
                        || msg.content_parts.is_some()
                        || msg.metadata.contains_key("empty_assistant_record");
                    // Only assign a synthetic group id when there's actual
                    // merge ambiguity to resolve (a message AND its own tool
                    // calls, or 2+ of this message's own tool calls) — a
                    // pure-text message with no tool calls, or a lone tool
                    // call with nothing else from the same `ChatMessage`,
                    // has nothing to disambiguate, so it keeps the exact
                    // historical byte shape (no `metadata` key at all).
                    let group_id: Option<String> = if let Some(real) = real_turn_id {
                        if used_group_ids.contains(real) {
                            // N2: this real turn_id was already used by an
                            // earlier (now-closed) `ChatMessage` in this same
                            // export — reusing it verbatim would let the
                            // reimport merge check recombine two originally
                            // separate messages (see the doc comment above).
                            let mut n = 1u64;
                            let mut candidate = format!("{real}~dup{n}");
                            while used_group_ids.contains(&candidate) {
                                n += 1;
                                candidate = format!("{real}~dup{n}");
                            }
                            Some(candidate)
                        } else {
                            Some(real.to_string())
                        }
                    } else if !msg.tool_calls().is_empty() {
                        // N2: skip past any id already used (e.g. a REAL
                        // turn_id that happens to look like `sc-grp-N`, or an
                        // id an earlier reused-real case landed on).
                        let mut candidate = format!("sc-grp-{next_group_id}");
                        next_group_id += 1;
                        while used_group_ids.contains(&candidate) {
                            candidate = format!("sc-grp-{next_group_id}");
                            next_group_id += 1;
                        }
                        Some(candidate)
                    } else {
                        None
                    };
                    if let Some(g) = &group_id {
                        used_group_ids.insert(g.clone());
                    }
                    if has_message_record {
                        self.push_codex_message(
                            out,
                            "assistant",
                            "output_text",
                            msg,
                            group_id.as_deref(),
                            None,
                        );
                    }
                    for tc in msg.tool_calls() {
                        let custom_tool_call = msg
                            .metadata
                            .get("codex_custom_tool_call_ids")
                            .and_then(|raw| serde_json::from_str::<Vec<String>>(raw).ok())
                            .is_some_and(|ids| ids.iter().any(|id| id == &tc.id));
                        if custom_tool_call {
                            let input = tc
                                .function
                                .parsed_arguments()
                                .unwrap_or_else(|_| Value::String(tc.function.arguments.clone()));
                            let mut payload = with_turn_id(
                                serde_json::json!({
                                    "type": "custom_tool_call",
                                    "name": tc.function.name,
                                    "input": input,
                                    "call_id": tc.id,
                                }),
                                group_id.as_deref(),
                            );
                            set_grok_message_extension(&mut payload, self.meta.source, msg);
                            push_jsonl(
                                out,
                                &codex_response_item(payload, msg_timestamp_or_synth(msg)),
                            );
                        } else if tc.function.name == "tool_search" {
                            tool_search_call_ids.insert(tc.id.clone());
                            let mut payload = with_turn_id(
                                serde_json::json!({
                                    "type": "tool_search_call",
                                    "arguments": tc.function.arguments,
                                    "call_id": tc.id,
                                }),
                                group_id.as_deref(),
                            );
                            set_grok_message_extension(&mut payload, self.meta.source, msg);
                            push_jsonl(
                                out,
                                &codex_response_item(payload, msg_timestamp_or_synth(msg)),
                            );
                        } else {
                            let mut payload = with_turn_id(
                                serde_json::json!({
                                    "type": "function_call",
                                    "name": tc.function.name,
                                    "arguments": tc.function.arguments,
                                    "call_id": tc.id,
                                }),
                                group_id.as_deref(),
                            );
                            set_grok_message_extension(&mut payload, self.meta.source, msg);
                            push_jsonl(
                                out,
                                &codex_response_item(payload, msg_timestamp_or_synth(msg)),
                            );
                        }
                    }
                    // PARITY-11: a genuinely reasoning-only turn (Claude
                    // `thinking`/`redacted_thinking` with no text, tool_use,
                    // or image — `push_claude_assistant`'s load-side fix for
                    // the ~21% of real assistant records that are exactly
                    // this shape) has no message record and no tool calls,
                    // so nothing above writes anything for it. This is
                    // DELIBERATE, not a residual gap: Codex's `reasoning`
                    // response_item is understood on import (see the
                    // `response_item`/`"reasoning"` arm above), but its
                    // real-native semantics is "the reasoning immediately
                    // BEFORE the next turn" — the reader attaches it to
                    // whatever response_item comes next, unconditionally.
                    // For a genuinely standalone Claude reasoning-only turn
                    // (no related turn follows in Codex's export at all),
                    // emitting one here would get silently misattributed as
                    // belonging to some later, unrelated turn instead —
                    // strictly worse than the current honest, accounted-for
                    // absence (thinking/redacted_thinking is provider-
                    // private and "not replayed across providers" by
                    // original design; the audit correctly classifies it
                    // `Coverage::Dropped`, not `Unmodeled`). See the
                    // PARITY-6/7 corpus test's `is_replayable` filter for
                    // why this doesn't count as a message-count regression.
                }
                Role::Tool
                    if msg
                        .tool_call_id
                        .as_deref()
                        .is_some_and(|id| tool_search_call_ids.contains(id)) =>
                {
                    let content = msg.content.clone().unwrap_or_default();
                    let tools =
                        serde_json::from_str::<Value>(&content).unwrap_or(Value::String(content));
                    let mut payload = serde_json::json!({
                        "type": "tool_search_output",
                        "call_id": msg.tool_call_id.clone().unwrap_or_default(),
                        "tools": tools,
                    });
                    set_grok_message_extension(&mut payload, self.meta.source, msg);
                    push_jsonl(
                        out,
                        &codex_response_item(payload, msg_timestamp_or_synth(msg)),
                    );
                }
                Role::Tool => {
                    let mut payload = serde_json::json!({
                        "type": "function_call_output",
                        "call_id": msg.tool_call_id.clone().unwrap_or_default(),
                        "output": codex_tool_output_text(msg),
                    });
                    set_grok_message_extension(&mut payload, self.meta.source, msg);
                    push_jsonl(
                        out,
                        &codex_response_item(payload, msg_timestamp_or_synth(msg)),
                    );
                }
            }
        }
    }

    /// A12 splice: replay the imported Codex `raw` prefix verbatim — every
    /// line, not just the `session_meta`/`turn_context` headers
    /// [`Self::to_codex_jsonl`] replays — overriding only
    /// `session_meta.payload.id` when `session_id` is `Some` (every other
    /// line, including `response_item`s the stock synthesis would otherwise
    /// rebuild from scratch, is untouched byte-for-byte). Then synthesizes
    /// `response_item` records only for the appended tail, via
    /// [`Self::write_codex_records`].
    pub(super) fn to_codex_jsonl_spliced(&self, session_id: Option<&str>) -> String {
        let (raw_prefix_len, message_prefix_len) = self.spliced_prefix_lens();

        let mut out = String::new();
        for line in &self.raw[..raw_prefix_len] {
            match session_id {
                Some(id) => {
                    let patched = serde_json::from_str::<Value>(line)
                        .ok()
                        .filter(|v| v.get("type").and_then(Value::as_str) == Some("session_meta"))
                        .map(|mut v| {
                            if let Some(payload) = v.get_mut("payload") {
                                payload["id"] = Value::String(id.to_string());
                            }
                            v.to_string()
                        });
                    out.push_str(patched.as_deref().unwrap_or(line));
                }
                None => out.push_str(line),
            }
            out.push('\n');
        }

        // N2 (spliced-path hardening): seed the tail's collision guard with
        // every group id the just-replayed RAW prefix already carries, so
        // `write_codex_records` never fabricates/reuses an id for the
        // appended tail that collides with one still open at the end of the
        // prefix (see that fn's doc comment, and
        // `collect_codex_group_ids_from_raw`'s).
        let mut seed_used_ids = collect_codex_group_ids_from_raw(&self.raw[..raw_prefix_len]);
        // Belt-and-suspenders: also union in the prefix `messages`' own
        // recorded `turn_id` metadata. In the ordinary case this is already
        // a subset of what the raw-line scan above found (the loader stamps
        // `metadata["turn_id"]` from the very same `payload.metadata.turn_id`
        // field the scan reads) — but scanning `messages` too costs nothing
        // and means this stays correct even if some future loader path ever
        // derives a message's `turn_id` by some means other than a literal
        // `payload.metadata.turn_id` copy.
        for msg in &self.messages[..message_prefix_len] {
            if let Some(tid) = msg.metadata.get("turn_id") {
                seed_used_ids.insert(tid.clone());
            }
        }
        self.write_codex_records(
            &mut out,
            &self.messages[message_prefix_len..],
            &seed_used_ids,
        );
        out
    }

    /// Build a Codex header from scratch (used when converting from another
    /// format, where no original Codex header exists to replay). Emits the
    /// fields Codex requires on `session_meta`.
    fn write_synthesized_codex_header(&self, out: &mut String) {
        let mut meta_payload = serde_json::json!({
            "id": self.meta.session_id.clone().unwrap_or_else(|| synth_uuid(0)),
            "timestamp": SYNTH_TS,
            "cwd": self.cwd_string(),
            "originator": "supercode",
            "cli_version": env!("CARGO_PKG_VERSION"),
            "source": "exec",
            "thread_source": "user",
            "model_provider": "openai",
        });
        if let Some(sp) = &self.meta.system_prompt {
            meta_payload["base_instructions"] = serde_json::json!({"text": sp});
        }
        // PARITY-10 dev/03: carry a captured Claude `fork-context-ref` (see
        // `capture_claude_meta`) through the Codex hop under a clearly
        // namespaced custom field — real Codex tooling ignores unknown
        // `session_meta.payload` keys, and `capture_codex_session_meta`
        // reads this same key back on import, so a Claude -> Codex -> Claude
        // round trip still reconstructs the original record instead of
        // silently losing the lineage note on the cross-format hop.
        if let Some(raw) = self.meta.lineage.get("claude_fork_context_ref_raw") {
            meta_payload["claude_fork_context_ref"] =
                serde_json::from_str(raw).unwrap_or_else(|_| Value::String(raw.clone()));
        }
        push_jsonl(
            out,
            &serde_json::json!({"timestamp": SYNTH_TS, "type": "session_meta", "payload": meta_payload}),
        );
        if let Some(model) = &self.meta.model {
            push_jsonl(
                out,
                &serde_json::json!({
                    "timestamp": SYNTH_TS,
                    "type": "turn_context",
                    "payload": {"model": model, "cwd": self.cwd_string()},
                }),
            );
        }
    }

    /// `turn_id`: see the PARITY-6/7 (and D3) comment on
    /// [`Self::write_codex_records`] — `Some` when the source message
    /// carries its own REAL `turn_id` (a native-Codex round-trip), or
    /// (assistant only) a synthetic disambiguation id when it owns tool
    /// calls needing merge disambiguation and has no real id of its own;
    /// `None` reproduces the exact historical shape (no `metadata` key at
    /// all).
    /// `extra_metadata`: PARITY-6 dev/02 — an additional `(key, value)`
    /// pair folded into `payload.metadata` alongside `turn_id` (used by the
    /// `Role::System` case in [`Self::write_codex_records`] to carry
    /// `claude_system_subtype`, so a Claude `<local-command-stdout>`-style
    /// system record's subtype survives the Claude -> Codex -> Claude round
    /// trip instead of only its text; `None` for every other caller,
    /// preserving the exact historical shape).
    fn push_codex_message(
        &self,
        out: &mut String,
        role: &str,
        text_type: &str,
        msg: &ChatMessage,
        turn_id: Option<&str>,
        extra_metadata: Option<(&str, &str)>,
    ) {
        let mut payload = with_turn_id(
            serde_json::json!({
                "type": "message",
                "role": role,
                "content": codex_message_content_blocks(text_type, msg),
            }),
            turn_id,
        );
        if let Some((k, v)) = extra_metadata {
            if payload.get("metadata").is_none() {
                payload["metadata"] = serde_json::json!({});
            }
            payload["metadata"][k] = serde_json::json!(v);
        }
        set_grok_message_extension(&mut payload, self.meta.source, msg);
        push_jsonl(
            out,
            &codex_response_item(payload, msg_timestamp_or_synth(msg)),
        );
    }
}

fn codex_response_item(payload: Value, ts: &str) -> Value {
    serde_json::json!({"timestamp": ts, "type": "response_item", "payload": payload})
}

/// Stamp `payload.metadata.turn_id` when `turn_id` is `Some` (PARITY-6/7,
/// see [`Session::write_codex_records`]); a no-op returning `payload`
/// untouched when `None`, so the historical byte shape is preserved for
/// every record that has no merge ambiguity to disambiguate.
fn with_turn_id(mut payload: Value, turn_id: Option<&str>) -> Value {
    if let Some(tid) = turn_id {
        payload["metadata"] = serde_json::json!({"turn_id": tid});
    }
    payload
}

/// Build a Codex `message` response_item's `content` block array from a
/// `ChatMessage` — the inverse of [`codex_extract_images`]/`extract_text_content`'s
/// parse. When `content_parts` is `None` this MUST reproduce the historical
/// single-block shape exactly (IX-5's overriding constraint: a text-only
/// message's export stays byte-identical) — only a multimodal message gets
/// one `{text_type}` block per non-empty text part plus one native Codex
/// `input_image` block (`{"type":"input_image","image_url":<data:URI or
/// URL>}` — the Responses-API-shaped image content Codex's own `input_text`/
/// `output_text` blocks already follow the family of) per `image_url` part.
fn codex_message_content_blocks(text_type: &str, msg: &ChatMessage) -> Value {
    match &msg.content_parts {
        Some(parts) => {
            let mut blocks = Vec::new();
            for p in parts {
                match p.get("type").and_then(Value::as_str) {
                    Some("text") => {
                        if let Some(t) = p.get("text").and_then(Value::as_str) {
                            if !t.is_empty() {
                                blocks.push(serde_json::json!({"type": text_type, "text": t}));
                            }
                        }
                    }
                    Some("image_url") => {
                        if let Some(url) = p
                            .get("image_url")
                            .and_then(|u| u.get("url"))
                            .and_then(Value::as_str)
                        {
                            blocks.push(serde_json::json!({
                                "type": "input_image",
                                "image_url": url,
                            }));
                        }
                    }
                    _ => {}
                }
            }
            Value::Array(blocks)
        }
        None => {
            let text = msg.content.clone().unwrap_or_default();
            Value::Array(vec![serde_json::json!({"type": text_type, "text": text})])
        }
    }
}

/// PARITY-11 (nested images, honest-residue side): a Codex
/// `function_call_output` response_item's `output` field is a BARE STRING
/// (real `codex-rs` protocol shape — unlike a `message` response_item, it has
/// no structured content array, so [`codex_message_content_blocks`]'s
/// `input_image` slot genuinely does not apply here). A nested image captured
/// off a Claude `tool_result` (`extract_tool_result_content`,
/// `content_parts`) therefore CANNOT be carried through this hop — but rather
/// than silently re-emitting the old bare `[image]` marker (indistinguishable
/// from a real, intentional annotation and impossible to tell apart from
/// "the data survived") or dropping it with zero trace, fold in an honest,
/// countable disclosure of exactly how many images were dropped and why —
/// same bracketed-note convention as [`UNCONVERTIBLE_IMAGE_MARKER`], applied
/// on the WRITE side instead of the read side. `content_parts` being `None`
/// (every pre-existing call site, and any tool result with no nested image)
/// reproduces the historical `msg.content` text byte-for-byte.
fn codex_tool_output_text(msg: &ChatMessage) -> String {
    let mut text = msg.content.clone().unwrap_or_default();
    if let Some(parts) = &msg.content_parts {
        let n = parts
            .iter()
            .filter(|p| p.get("type").and_then(Value::as_str) == Some("image_url"))
            .count();
        if n > 0 {
            if !text.is_empty() {
                text.push('\n');
            }
            text.push_str(&format!(
                "[image: {n} nested image(s) dropped — codex tool output has no \
                 structured content slot to carry them]"
            ));
        }
    }
    text
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn bounded_codex_file_expands_past_tool_noise_and_loads_real_earlier_history() {
        let nonce = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .unwrap()
            .as_nanos();
        let path = std::env::temp_dir().join(format!(
            "supercode-display-history-{}-{nonce}.jsonl",
            std::process::id()
        ));
        let user = |text: &str| {
            format!(
                r#"{{"type":"response_item","payload":{{"type":"message","role":"user","content":[{{"type":"input_text","text":"{text}"}}]}}}}"#,
            )
        };
        let assistant = |index: usize| {
            format!(
                r#"{{"type":"response_item","payload":{{"type":"message","role":"assistant","content":[{{"type":"output_text","text":"answer-{index}"}}]}}}}"#,
            )
        };
        let mut lines = vec![
            r#"{"type":"session_meta","payload":{"id":"session-1"}}"#.to_string(),
            user("earlier prompt"),
            format!(
                r#"{{"type":"event_msg","payload":{{"type":"token_count","noise":"{}"}}}}"#,
                "x".repeat(5 * 1024 * 1024)
            ),
            user("latest prompt"),
        ];
        lines.extend((0..130).map(assistant));
        std::fs::write(&path, format!("{}\n", lines.join("\n"))).unwrap();

        let initial = Session::load_display_view(&path, Fidelity::Semantic, 120).unwrap();
        let expanded = Session::load_display_view(&path, Fidelity::Semantic, 240).unwrap();
        std::fs::remove_file(&path).unwrap();

        let initial_users = initial
            .messages
            .iter()
            .filter(|message| message.role == Role::User)
            .filter_map(|message| message.content.as_deref())
            .collect::<Vec<_>>();
        assert_eq!(initial.messages.len(), 120);
        assert_eq!(initial_users, ["earlier prompt", "latest prompt"]);
        assert!(
            initial.imported_message_count.unwrap() > initial.messages.len(),
            "a bounded initial page must truthfully report earlier history"
        );
        assert_eq!(expanded.messages.len(), 132);
        assert_eq!(expanded.imported_message_count, Some(132));
    }

    #[test]
    fn bounded_codex_display_history_reports_the_unbounded_message_total() {
        let jsonl = (0..6)
            .map(|index| {
                let role = if index % 2 == 0 { "user" } else { "assistant" };
                format!(
                    r#"{{"type":"response_item","payload":{{"type":"message","role":"{role}","content":[{{"type":"input_text","text":"message-{index}"}}]}}}}"#,
                )
            })
            .collect::<Vec<_>>()
            .join("\n");

        let session = Session::from_codex_display_str(&jsonl, 2).unwrap();

        assert_eq!(session.messages.len(), 2);
        assert_eq!(session.imported_message_count, Some(6));
    }
}