supercode_interchange/session/

pi.rs

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

use super::*;

impl Session {
    /// Load a Pi session from a file.
    pub fn from_pi(path: impl AsRef<Path>) -> Result<Session> {
        Self::from_pi_str(&std::fs::read_to_string(path.as_ref())?)
    }

    /// Parse a Pi session (`docs/interop/opencode-pi-spec.md` §1.1,
    /// `docs/interop/research/pi-fields.md`) from an in-memory JSONL string.
    ///
    /// Line 1 is the `session` header; every other line is one `SessionEntry`
    /// in a tree keyed by `id`/`parentId` — file order is append order, not
    /// tree order. `raw` captures every line verbatim (byte-lossless T1,
    /// exactly like Claude Code/Codex). `messages` is the **active path
    /// only**: pi's own leaf rule is "the last entry in file order"
    /// (`pi-fields.md` `sm:897`), so this walks `parentId` from there back to
    /// the root and linearizes root→leaf. Non-active branches, `label`s, and
    /// state records (`thinking_level_change`/`model_change`/`custom`/
    /// `session_info`) are never visited by that walk — they survive in
    /// `raw` only, pi's defining residue (§1.1).
    ///
    /// `message.role` is an OPEN union upstream (§1.1 S6): a role outside the
    /// five modeled here (`user`/`assistant`/`toolResult`/`bashExecution`/
    /// `custom`) produces no canonical message — raw-only survival, never a
    /// panic — and the Pi corpus audit turns that into a
    /// visible coverage failure rather than a silent drop.
    ///
    /// Same fail-loud discipline applies to `ImageContent` blocks
    /// (`user`/`toolResult`/`custom*` content, see `pi_image_shape`): the
    /// assumed `{mimeType, data}` shape is UNVERIFIED against real pi output
    /// (`pi-fields.md` doesn't enumerate `ImageContent`'s own fields, only
    /// cites the containing union) — a follow-up TR tracks confirming it
    /// against a real corpus. Until then, an image block that doesn't match
    /// that shape never gets silently synthesized as an empty/corrupt
    /// `image_url` part; the containing message survives in `raw` only and
    /// trips `crate::audit::Corpus::Pi`'s `message/UnknownImageShape` bucket.
    pub fn from_pi_str(jsonl: &str) -> Result<Session> {
        Self::from_pi_v3_dialect(jsonl, SessionSource::Pi, false)
    }

    pub(super) fn from_pi_v3_dialect(
        jsonl: &str,
        source: SessionSource,
        openclaw: bool,
    ) -> Result<Session> {
        let mut meta = SessionMeta::new(source);
        // IX-1: `raw` is captured STRICT-VERBATIM — separate from the
        // blank-skipping PARSE walk (`lines_v`) below, which must keep
        // skipping blank/whitespace-only lines when it looks for `SessionEntry`
        // records (a blank line is never a record, on either view).
        let (raw_lines, raw_trailing_newline) = split_lines_verbatim(jsonl);
        let raw: Vec<String> = raw_lines.iter().map(|l| l.to_string()).collect();
        let non_empty_line_count = non_empty_lines(jsonl).count();
        let lines_v: Vec<Value> = non_empty_lines(jsonl)
            .filter_map(|l| serde_json::from_str(l).ok())
            .collect();
        // PARITY-15: every line that failed to even deserialize as JSON at
        // all (never mind whether it then parsed as a recognized
        // `SessionEntry` shape) — see `from_claude_code_str`'s identical
        // counter.
        let parse_error_lines = non_empty_line_count.saturating_sub(lines_v.len());

        if let Some(header) = lines_v.first() {
            capture_pi_header(header, &mut meta)?;
            if openclaw {
                openclaw_capture_header_nouns(header, &mut meta);
            }
        }

        // Every non-header entry that parses as an object carrying an `id`.
        // (A line that fails to parse, or a header re-parsed as an entry,
        // simply never enters `by_id` — it survives in `raw` only, exactly
        // like a malformed/non-conversational line in the other loaders.)
        struct PiEntry {
            id: String,
            parent_id: Option<String>,
            value: Value,
        }
        let mut entries: Vec<PiEntry> = Vec::new();
        let mut by_id: HashMap<String, usize> = HashMap::new();
        for v in lines_v.iter().skip(1) {
            let Some(id) = v.get("id").and_then(Value::as_str) else {
                continue;
            };
            let parent_id = v
                .get("parentId")
                .and_then(Value::as_str)
                .map(str::to_string);
            by_id.insert(id.to_string(), entries.len());
            entries.push(PiEntry {
                id: id.to_string(),
                parent_id,
                value: v.clone(),
            });
        }

        if entries.is_empty() {
            return Ok(Session {
                meta,
                messages: Vec::new(),
                subagents: Vec::new(),
                raw,
                raw_trailing_newline,
                imported_message_count: Some(0),
                // Pi is line-oriented: `raw` is split directly out of the
                // source text (strict-verbatim, IX-1), even for this
                // no-entries early return.
                raw_is_verbatim: true,
                parse_error_lines,
                load_residue: Vec::new(),
            });
        }

        // Leaf = the last entry in file order (pi's own rule, `sm:897`), NOT
        // necessarily a `message` entry — a trailing `label`/`session_info`
        // still anchors the walk correctly since the walk just follows
        // `parentId` regardless of the leaf's own type.
        //
        // OpenClaw dialect: a `type:"leaf"` entry REDIRECTS the anchor to its
        // `targetId` (last one wins); with none — or a dangling/null target —
        // the default rule applies, skipping trailing `leaf` entries
        // themselves and `appendMode:"side"` entries, which never anchor.
        let default_leaf_idx = if openclaw {
            entries
                .iter()
                .rposition(|entry| {
                    entry.value.get("type").and_then(Value::as_str) != Some("leaf")
                        && entry.value.get("appendMode").and_then(Value::as_str) != Some("side")
                })
                .unwrap_or(entries.len() - 1)
        } else {
            entries.len() - 1
        };
        let leaf_idx = if openclaw {
            entries
                .iter()
                .rev()
                .find(|entry| entry.value.get("type").and_then(Value::as_str) == Some("leaf"))
                .and_then(|redirect| {
                    redirect
                        .value
                        .get("targetId")
                        .and_then(Value::as_str)
                        .and_then(|target| by_id.get(target).copied())
                })
                .unwrap_or(default_leaf_idx)
        } else {
            default_leaf_idx
        };
        let mut chain_rev: Vec<usize> = Vec::new();
        let mut cur: Option<String> = Some(entries[leaf_idx].id.clone());
        let mut guard = 0usize;
        while let Some(id) = cur {
            let Some(&idx) = by_id.get(&id) else { break };
            chain_rev.push(idx);
            cur = entries[idx].parent_id.clone();
            guard += 1;
            if guard > entries.len() + 1 {
                break; // cycle guard — malformed parentId chain
            }
        }
        chain_rev.reverse();
        let active = chain_rev; // indices into `entries`, root..leaf order

        let pos_in_active: HashMap<&str, usize> = active
            .iter()
            .enumerate()
            .map(|(pos, &idx)| (entries[idx].id.as_str(), pos))
            .collect();

        // First pass: compaction discipline (§2.1 S3) — every message from an
        // entry before the LATEST `firstKeptEntryId` on the active path is
        // excluded from replay (`compacted_out`), mirroring pi's own
        // `buildContextEntries` slice (`sm:414-450`).
        let mut kept_from_pos = 0usize;
        for &idx in &active {
            let e = &entries[idx];
            if e.value.get("type").and_then(Value::as_str) == Some("compaction") {
                if let Some(fk) = e.value.get("firstKeptEntryId").and_then(Value::as_str) {
                    if let Some(&p) = pos_in_active.get(fk) {
                        kept_from_pos = kept_from_pos.max(p);
                    }
                }
            }
        }

        let mut messages = Vec::new();
        let mut current_model: Option<String> = None;
        for (pos, &idx) in active.iter().enumerate() {
            let e = &entries[idx];
            let v = &e.value;
            let entry_ts = v
                .get("timestamp")
                .and_then(Value::as_str)
                .map(str::to_string);
            let before = messages.len();
            match v.get("type").and_then(Value::as_str) {
                Some("message") => {
                    let msg_v = v.get("message").cloned().unwrap_or(Value::Null);
                    match msg_v.get("role").and_then(Value::as_str) {
                        Some("user") => push_pi_user(&msg_v, &mut messages),
                        Some("assistant") => {
                            push_pi_assistant(&msg_v, &mut messages);
                            if let Some(m) = msg_v.get("model").and_then(Value::as_str) {
                                current_model = Some(m.to_string());
                            }
                        }
                        Some("toolResult") => push_pi_tool_result(&msg_v, &mut messages),
                        Some("bashExecution") => push_pi_bash(&msg_v, &mut messages),
                        Some("custom") => push_pi_custom_common(&msg_v, &mut messages),
                        // OPEN UNION (S6): any other role — raw-only survival.
                        _ => {}
                    }
                }
                Some("custom_message") => push_pi_custom_common(v, &mut messages),
                Some("compaction") => push_pi_compaction(v, &mut messages),
                Some("branch_summary") => push_pi_branch_summary(v, &mut messages),
                Some("model_change") => {
                    if let Some(m) = v.get("modelId").and_then(Value::as_str) {
                        current_model = Some(m.to_string());
                    }
                }
                Some("session_info") => {
                    if let Some(name) = v.get("name").and_then(Value::as_str) {
                        if !name.is_empty() {
                            meta.lineage
                                .insert("session_name".to_string(), name.to_string());
                        }
                    }
                }
                // thinking_level_change, custom (entry-level state), label —
                // no clean home, raw-only (§2.3).
                _ => {}
            }
            let is_summary = matches!(
                v.get("type").and_then(Value::as_str),
                Some("compaction") | Some("branch_summary")
            );
            for m in &mut messages[before..] {
                m.metadata.insert("pi_entry_id".to_string(), e.id.clone());
                if openclaw {
                    // Vendor metadata (`message.__openclaw.*`) — preserved as
                    // provenance, never interpreted. Session-key/delegate
                    // references inside it stay inert strings (mirror, not
                    // recursion).
                    if let Some(vendor) = v
                        .get("message")
                        .and_then(|mm| mm.get("__openclaw"))
                        .and_then(Value::as_object)
                    {
                        for (key, value) in vendor {
                            let rendered = match value {
                                Value::String(text) => text.clone(),
                                other => other.to_string(),
                            };
                            m.metadata.insert(format!("openclaw_{key}"), rendered);
                        }
                    }
                }
                if let Some(p) = &e.parent_id {
                    m.metadata.insert("pi_parent_id".to_string(), p.clone());
                }
                if let Some(ts) = &entry_ts {
                    m.metadata
                        .entry("timestamp".to_string())
                        .or_insert_with(|| ts.clone());
                }
                // WAVE-2 item 1 fallback: the entry-level `timestamp` above
                // is pi's authoritative, always-monotonic-in-file-order
                // wall-clock (mandatory on every entry) and wins whenever
                // present. The nested `message.timestamp` (unix-ms) is only
                // reached here — via `entry(...).or_insert_with`, so it
                // never overwrites the entry-level value — in the rare case
                // an entry lacks its own `timestamp`. This intentionally
                // does NOT prefer the msg-level field even though it LOOKS
                // more precise: unlike the entry-level timestamp, it is not
                // guaranteed monotonic with this loader's root->leaf
                // linearization (e.g. a rewound-branch entry can carry an
                // earlier msg-level clock reading than its file-order
                // neighbors), and OpenCode's own loader re-sorts messages by
                // this canonical timestamp — a non-monotonic source would
                // silently scramble replay order on a pi->opencode hop.
                if let Some(ms) = v
                    .get("message")
                    .and_then(|mm| mm.get("timestamp"))
                    .and_then(Value::as_u64)
                {
                    m.metadata
                        .entry("timestamp".to_string())
                        .or_insert_with(|| crate::sidecar::ms_to_rfc3339(ms as i64));
                }
                // A compaction/branch-summary message IS the retained marker
                // — never mark it excluded, regardless of its own position.
                if !is_summary && pos < kept_from_pos {
                    m.metadata
                        .insert("compacted_out".to_string(), "true".to_string());
                }
            }
            restore_single_grok_message(v, &mut messages[before..]);
            for message in &mut messages[before..] {
                restore_tool_outcome_extension(v, message);
            }
        }

        meta.model = current_model;
        ensure_tool_results_paired(&mut messages);
        // Each message's creating record, by the entry id the loader stamped on it.
        let record_of_entry: HashMap<String, usize> = raw_lines
            .iter()
            .enumerate()
            .filter_map(|(index, line)| {
                let value: Value = serde_json::from_str(line).ok()?;
                Some((value.get("id")?.as_str()?.to_string(), index))
            })
            .collect();
        meta.message_records = messages
            .iter()
            .map(|message| {
                message
                    .metadata
                    .get("pi_entry_id")
                    .and_then(|id| record_of_entry.get(id).copied())
            })
            .collect();
        let imported_message_count = Some(messages.len());
        Ok(Session {
            meta,
            messages,
            subagents: Vec::new(),
            raw,
            raw_trailing_newline,
            imported_message_count,
            // Pi 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(),
        })
    }
}

// ---- Pi ---------------------------------------------------------------

fn capture_pi_header(v: &Value, meta: &mut SessionMeta) -> Result<()> {
    restore_codex_provenance_from_top_level(v, meta)?;
    if let Some(id) = v.get("id").and_then(Value::as_str) {
        meta.session_id = Some(id.to_string());
    }
    if let Some(cwd) = v.get("cwd").and_then(Value::as_str) {
        meta.cwd = Some(PathBuf::from(cwd));
    }
    // Absent `version` means a pre-v3 file (`pi-fields.md` §1: "absent = v1").
    let version = v
        .get("version")
        .and_then(Value::as_u64)
        .map(|n| n.to_string())
        .unwrap_or_else(|| "1".to_string());
    meta.lineage.insert("pi_version".to_string(), version);
    if let Some(ts) = v.get("timestamp").and_then(Value::as_str) {
        meta.lineage
            .insert("created_at".to_string(), ts.to_string());
    }
    if let Some(ps) = v.get("parentSession").and_then(Value::as_str) {
        meta.lineage
            .insert("parent_session_path".to_string(), ps.to_string());
    }
    // D7: the other half of `push_pi_header`'s passthrough — restores a
    // captured Claude `fork-context-ref` so a Claude -> Pi -> Claude round
    // trip reconstructs the original record (mirrors
    // `capture_codex_session_meta`'s identical `claude_fork_context_ref`
    // restore for the Codex hop).
    if !meta.lineage.contains_key("claude_fork_context_ref_raw") {
        if let Some(v) = v.get("claude_fork_context_ref") {
            meta.lineage
                .insert("claude_fork_context_ref_raw".to_string(), v.to_string());
        }
    }
    Ok(())
}

/// Validate one `ImageContent` block's `{mimeType, data}` shape and return
/// `(mime, data)` when it looks like a real image payload.
///
/// **This shape is UNVERIFIED against real pi output**: `pi-fields.md` cites
/// `ai:316-350` for the `ImageContent` content-block union but does not
/// enumerate `ImageContent`'s own fields (only `TextContent`/`ThinkingContent`/
/// `ToolCall` are itemized there) — `{mimeType, data}` (mirroring the OpenAI/
/// Anthropic multimodal wire shape) is this loader's best guess, not a
/// frozen-spec fact. A follow-up TR tracks confirming/correcting this shape
/// against a real pi corpus. Until then this function VALIDATES rather than
/// assumes: both fields must be present, non-empty strings, and `data` must
/// look like base64 (only the base64 alphabet, incl. `=` padding) — anything
/// else is an unknown/unexpected image shape, and the caller must route the
/// whole message to raw-only survival (S6-style fail loud) instead of
/// silently synthesizing a corrupt/empty `image_url` part.
fn pi_image_shape(item: &Value) -> Option<(String, String)> {
    let mime = item.get("mimeType").and_then(Value::as_str)?;
    let data = item.get("data").and_then(Value::as_str)?;
    if mime.is_empty() || data.is_empty() {
        return None;
    }
    if !data
        .bytes()
        .all(|b| b.is_ascii_alphanumeric() || matches!(b, b'+' | b'/' | b'='))
    {
        return None;
    }
    Some((mime.to_string(), data.to_string()))
}

/// True if `content` (a pi content value: bare string or
/// `(TextContent|ImageContent)[]`) contains at least one `type:"image"` block
/// that does not match [`pi_image_shape`] — shared by the loader (which
/// routes such a message to raw-only survival, never a synthesized-empty
/// part) and the coverage audit (`audit::Corpus::Pi`), which scores it as
/// `message/UnknownImageShape` instead of silently `Normalized`, so the shape
/// mismatch surfaces as a coverage FAILURE rather than vanishing.
#[doc(hidden)]
pub fn pi_content_has_unknown_image_shape(content: Option<&Value>) -> bool {
    let Some(Value::Array(items)) = content else {
        return false;
    };
    items.iter().any(|item| {
        item.get("type").and_then(Value::as_str) == Some("image") && pi_image_shape(item).is_none()
    })
}

/// Split a pi `(TextContent|ImageContent)[]` (or bare string) content value
/// into concatenated text plus, when a WELL-FORMED image block is present,
/// the full `content_parts` array (leading text block + one `image_url` part
/// per image, its `data:` URI carrying the exact `mimeType`/`data` bytes pi
/// stored) — shared by `user`/`toolResult`/`custom*` content, which all use
/// the identical union (`pi-fields.md` §3a/§3c/§3e).
///
/// Returns `(text, parts, unknown_image_shape)`. When an image block does NOT
/// match [`pi_image_shape`] (missing/empty `data`/`mimeType`, or a `data`
/// value that isn't recognizable base64), this NEVER synthesizes an empty/
/// corrupt `image_url` part — it reports `unknown_image_shape = true` and
/// every caller must treat that as raw-only survival for the whole message
/// (mirroring the unknown-`message.role` rule, S6), so a shape this loader
/// guessed wrong fails loud instead of silently dropping/corrupting the
/// image.
fn pi_content_to_text_and_parts(content: Option<&Value>) -> (String, Option<Vec<Value>>, bool) {
    match content {
        Some(Value::String(s)) => (s.clone(), None, false),
        Some(Value::Array(items)) => {
            let mut text = String::new();
            let mut parts: Vec<Value> = Vec::new();
            let mut has_image = false;
            let mut unknown_image_shape = false;
            for item in items {
                match item.get("type").and_then(Value::as_str) {
                    Some("text") => {
                        if let Some(t) = item.get("text").and_then(Value::as_str) {
                            push_str_field(&mut text, t);
                        }
                    }
                    Some("image") => {
                        has_image = true;
                        match pi_image_shape(item) {
                            Some((mime, data)) => {
                                parts.push(serde_json::json!({
                                    "type": "image_url",
                                    "image_url": {"url": format!("data:{mime};base64,{data}")},
                                }));
                            }
                            None => unknown_image_shape = true,
                        }
                    }
                    _ => {}
                }
            }
            if unknown_image_shape {
                // Never synthesize an empty/corrupt part for a shape we
                // don't recognize — raw-only survival for the whole message;
                // the coverage guard is what turns this into a visible
                // failure (S6-style).
                return (String::new(), None, true);
            }
            if has_image {
                if !text.trim().is_empty() {
                    parts.insert(0, serde_json::json!({"type": "text", "text": text.clone()}));
                }
                (text, Some(parts), false)
            } else {
                (text, None, false)
            }
        }
        _ => (String::new(), None, false),
    }
}

fn push_pi_user(msg_v: &Value, out: &mut Vec<ChatMessage>) {
    let (text, parts, unknown_image_shape) = pi_content_to_text_and_parts(msg_v.get("content"));
    // Unrecognized image shape (S6-style fail loud, FIX #2): raw-only
    // survival, never a synthesized-empty part. `audit::Corpus::Pi`'s
    // `message/UnknownImageShape` bucket is what turns this into a visible
    // coverage failure.
    if unknown_image_shape {
        return;
    }
    if text.trim().is_empty() && parts.is_none() {
        return;
    }
    let mut msg = match parts {
        Some(parts) => ChatMessage {
            role: Role::User,
            content: None,
            content_parts: Some(parts),
            tool_calls: None,
            tool_call_id: None,
            name: None,
            metadata: Default::default(),
        },
        None => ChatMessage::user(text),
    };
    // WAVE-2 fidelity fix: pi's message-level unix-ms clock
    // (`message.timestamp`) is a DISTINCT field from the canonical
    // entry-level ISO `metadata["timestamp"]` WAVE-2 item 1 wires — the two
    // carry genuinely different values in real corpora (the fixture's are
    // ~6 months apart). Preserve it separately so it isn't silently lost for
    // every pi session; see `msg_pi_native_timestamp_ms` (the pi writer's
    // native round-trip consumer) and the INHERENT residue note on
    // `pi_dropped_keys_cross` in `interop_fidelity_matrix.rs`.
    if let Some(ts) = msg_v.get("timestamp").and_then(Value::as_u64) {
        msg.metadata
            .insert("pi_msg_timestamp".to_string(), ts.to_string());
    }
    out.push(msg);
}

fn push_pi_assistant(msg_v: &Value, out: &mut Vec<ChatMessage>) {
    let mut text = String::new();
    let mut calls: Vec<ToolCall> = Vec::new();
    let mut thinking = String::new();
    let mut thinking_seen = false;
    let mut thinking_sig: Option<String> = None;
    let mut thinking_redacted = false;
    let mut text_sig: Option<String> = None;
    let mut thought_sig: Option<String> = None;

    if let Some(Value::Array(blocks)) = msg_v.get("content") {
        for b in blocks {
            match b.get("type").and_then(Value::as_str) {
                Some("text") => {
                    if let Some(t) = b.get("text").and_then(Value::as_str) {
                        push_str_field(&mut text, t);
                    }
                    if let Some(sig) = b.get("textSignature") {
                        text_sig = Some(match sig {
                            Value::String(s) => s.clone(),
                            other => other.to_string(),
                        });
                    }
                }
                Some("thinking") => {
                    thinking_seen = true;
                    if let Some(t) = b.get("thinking").and_then(Value::as_str) {
                        push_str_field(&mut thinking, t);
                    }
                    if let Some(sig) = b.get("thinkingSignature").and_then(Value::as_str) {
                        thinking_sig = Some(sig.to_string());
                    }
                    if b.get("redacted").and_then(Value::as_bool) == Some(true) {
                        thinking_redacted = true;
                    }
                }
                Some("toolCall") => {
                    let id = b.get("id").and_then(Value::as_str).unwrap_or_default();
                    let name = b.get("name").and_then(Value::as_str).unwrap_or_default();
                    // `arguments` is a JSON OBJECT on pi's wire, not a string
                    // (`pi-fields.md` §3b open question 4) — serialize to the
                    // string `FunctionCall::arguments` expects.
                    let args = b
                        .get("arguments")
                        .cloned()
                        .unwrap_or_else(|| Value::Object(Default::default()));
                    calls.push(function_call(id, name, args.to_string()));
                    if let Some(sig) = b.get("thoughtSignature").and_then(Value::as_str) {
                        thought_sig = Some(sig.to_string());
                    }
                }
                _ => {}
            }
        }
    }

    let before = out.len();
    push_assistant(out, text, calls);
    // A recognized native assistant entry remains transcript state even
    // when its content array is empty, except Pi's explicit empty error
    // response: that record has no replayable content and is established
    // raw-only residue (`pi_real_corpus_error_retry`). Preserve empty
    // non-error turns and Pi's standalone thinking-block shape.
    let is_empty_error =
        !thinking_seen && msg_v.get("stopReason").and_then(Value::as_str) == Some("error");
    if out.len() == before && !is_empty_error {
        let mut empty = ChatMessage {
            role: Role::Assistant,
            content: None,
            content_parts: None,
            tool_calls: None,
            tool_call_id: None,
            name: None,
            metadata: Default::default(),
        };
        if !thinking_seen {
            empty
                .metadata
                .insert("empty_assistant_record".to_string(), "true".to_string());
        }
        out.push(empty);
    }
    if out.len() > before {
        let msg = out.last_mut().expect("just pushed");
        if thinking_seen {
            msg.metadata.insert("thinking".to_string(), thinking);
        }
        if let Some(s) = thinking_sig {
            msg.metadata.insert("thinking_signature".to_string(), s);
        }
        if thinking_redacted {
            msg.metadata
                .insert("pi_thinking_redacted".to_string(), "true".to_string());
        }
        if let Some(s) = text_sig {
            msg.metadata.insert("pi_text_signature".to_string(), s);
        }
        if let Some(s) = thought_sig {
            msg.metadata.insert("pi_thought_signature".to_string(), s);
        }
        for (key, field) in [
            ("pi_api", "api"),
            ("pi_provider", "provider"),
            ("pi_response_model", "responseModel"),
            ("pi_response_id", "responseId"),
            ("pi_stop_reason", "stopReason"),
            ("pi_error_message", "errorMessage"),
        ] {
            if let Some(s) = msg_v.get(field).and_then(Value::as_str) {
                msg.metadata.insert(key.to_string(), s.to_string());
            }
        }
        if let Some(diag) = msg_v.get("diagnostics") {
            if !diag.is_null() {
                msg.metadata
                    .insert("pi_diagnostics".to_string(), diag.to_string());
            }
        }
        if let Some(usage) = msg_v.get("usage") {
            if !usage.is_null() {
                msg.metadata
                    .insert("pi_usage".to_string(), usage.to_string());
            }
        }
        // WAVE-2 fidelity fix: preserve pi's message-level unix-ms clock
        // separately from the canonical entry-level ISO `timestamp` — see
        // `push_pi_user`.
        if let Some(ts) = msg_v.get("timestamp").and_then(Value::as_u64) {
            msg.metadata
                .insert("pi_msg_timestamp".to_string(), ts.to_string());
        }
    }
}

fn push_pi_tool_result(msg_v: &Value, out: &mut Vec<ChatMessage>) {
    let id = msg_v
        .get("toolCallId")
        .and_then(Value::as_str)
        .unwrap_or_default();
    let name = msg_v
        .get("toolName")
        .and_then(Value::as_str)
        .unwrap_or_default();
    let (text, parts, unknown_image_shape) = pi_content_to_text_and_parts(msg_v.get("content"));
    // Unrecognized image shape (S6-style fail loud, FIX #2): raw-only
    // survival, never a synthesized-empty part. Dropping the toolResult
    // message here leaves its `toolCallId` unanswered, which
    // `ensure_tool_results_paired` already turns into a visible
    // "[no tool result recorded — turn interrupted]" placeholder — a loud
    // failure mode, not a silent one.
    if unknown_image_shape {
        return;
    }
    let mut msg = ChatMessage {
        role: Role::Tool,
        content: Some(text),
        content_parts: parts,
        tool_calls: None,
        tool_call_id: Some(id.to_string()),
        name: Some(name.to_string()),
        metadata: Default::default(),
    };
    if let Some(details) = msg_v.get("details") {
        if !details.is_null() {
            msg.metadata
                .insert("pi_tool_details".to_string(), details.to_string());
        }
    }
    let is_error = msg_v
        .get("isError")
        .and_then(Value::as_bool)
        .unwrap_or(false);
    msg.metadata
        .insert("pi_is_error".to_string(), is_error.to_string());
    if is_error {
        crate::mark_tool_error(&mut msg);
    }
    // WAVE-2 fidelity fix: preserve pi's message-level unix-ms clock
    // separately from the canonical entry-level ISO `timestamp` — see
    // `push_pi_user`.
    if let Some(ts) = msg_v.get("timestamp").and_then(Value::as_u64) {
        msg.metadata
            .insert("pi_msg_timestamp".to_string(), ts.to_string());
    }
    out.push(msg);
}

/// Render a pi `bashExecution` message (`!`/`!!` shell escape) to the text
/// pi itself sends the model, mirroring `bashExecutionToText`
/// (`pi-fields.md` §3d, `msg:82-98`). The exact upstream string constants
/// aren't reproduced in the frozen research doc (only cited by file:line),
/// so this is a faithful, clearly-labeled reconstruction — every structured
/// field is additionally preserved verbatim in `metadata`/`raw` regardless.
fn push_pi_bash(msg_v: &Value, out: &mut Vec<ChatMessage>) {
    let command = msg_v.get("command").and_then(Value::as_str).unwrap_or("");
    let output = msg_v.get("output").and_then(Value::as_str).unwrap_or("");
    let exit_code = msg_v.get("exitCode").and_then(Value::as_i64);
    let cancelled = msg_v
        .get("cancelled")
        .and_then(Value::as_bool)
        .unwrap_or(false);
    let truncated = msg_v
        .get("truncated")
        .and_then(Value::as_bool)
        .unwrap_or(false);

    let mut text = format!("$ {command}\n{output}");
    if let Some(code) = exit_code {
        if code != 0 {
            text.push_str(&format!("\n[exit code: {code}]"));
        }
    }
    if cancelled {
        text.push_str("\n[cancelled]");
    }
    if truncated {
        text.push_str("\n[truncated]");
    }

    let mut msg = ChatMessage::user(text);
    msg.metadata
        .insert("pi_bash_command".to_string(), command.to_string());
    msg.metadata
        .insert("pi_bash_output".to_string(), output.to_string());
    if let Some(code) = exit_code {
        msg.metadata
            .insert("pi_bash_exit_code".to_string(), code.to_string());
    }
    msg.metadata
        .insert("pi_bash_cancelled".to_string(), cancelled.to_string());
    msg.metadata
        .insert("pi_bash_truncated".to_string(), truncated.to_string());
    if let Some(p) = msg_v.get("fullOutputPath").and_then(Value::as_str) {
        msg.metadata
            .insert("pi_bash_full_output_path".to_string(), p.to_string());
    }
    // `!!` — hidden from the model context; honored by `is_replay_excluded`
    // on every writer, not just pi's own (§2.2).
    if msg_v.get("excludeFromContext").and_then(Value::as_bool) == Some(true) {
        msg.metadata
            .insert("pi_exclude_from_context".to_string(), "true".to_string());
    }
    // WAVE-2 fidelity fix: preserve pi's message-level unix-ms clock
    // separately from the canonical entry-level ISO `timestamp` — see
    // `push_pi_user`.
    if let Some(ts) = msg_v.get("timestamp").and_then(Value::as_u64) {
        msg.metadata
            .insert("pi_msg_timestamp".to_string(), ts.to_string());
    }
    out.push(msg);
}

/// B4: the `customType` marker `write_pi_entries`'s `Role::System` arm
/// stamps on a re-materialized content-bearing Claude `system` record (see
/// that arm's doc comment). Namespaced (`supercode_`-prefixed) so it can
/// never collide with a real pi `CustomMessage.customType` — pi's own
/// hook-injected custom types are hook/extension names (e.g. `hookMessage`
/// migration targets), never this literal string.
const PI_CLAUDE_SYSTEM_CUSTOM_TYPE: &str = "supercode_claude_system";

/// Shared mapping for pi's `role:"custom"` message (§3e) and top-level
/// `custom_message` entries (§9) — both enter context as a `User` message
/// with the same `customType`/`display`/`details` residue.
///
/// B4 exception: when `customType` is [`PI_CLAUDE_SYSTEM_CUSTOM_TYPE`] (our
/// own marker — see `write_pi_entries`'s `Role::System` arm), this is
/// actually a re-materialized content-bearing Claude `system` record round-
/// tripping through pi, not a genuine pi extension message — restore
/// `Role::System` + `metadata["systemSubtype"]` (from `details.
/// claude_system_subtype`, falling back to `local_command` — still one of
/// `push_claude_system`'s own keep subtypes — exactly like
/// `write_codex_records`'s Codex-leg fallback) instead of the generic
/// `Role::User` path below, so a Claude -> Pi -> Claude round trip restores
/// the exact original role, not just the text.
fn push_pi_custom_common(v: &Value, out: &mut Vec<ChatMessage>) {
    if v.get("customType").and_then(Value::as_str) == Some(PI_CLAUDE_SYSTEM_CUSTOM_TYPE) {
        let content = v.get("content").and_then(Value::as_str).unwrap_or("");
        if content.trim().is_empty() {
            return;
        }
        let subtype = v
            .get("details")
            .and_then(|d| d.get("claude_system_subtype"))
            .and_then(Value::as_str)
            .unwrap_or("local_command");
        out.push(ChatMessage::system(content.to_string()).with_meta("systemSubtype", subtype));
        return;
    }
    let (text, parts, unknown_image_shape) = pi_content_to_text_and_parts(v.get("content"));
    // Unrecognized image shape (S6-style fail loud, FIX #2): raw-only
    // survival, never a synthesized-empty part.
    if unknown_image_shape {
        return;
    }
    if text.trim().is_empty() && parts.is_none() {
        return;
    }
    let mut msg = match parts {
        Some(parts) => ChatMessage {
            role: Role::User,
            content: None,
            content_parts: Some(parts),
            tool_calls: None,
            tool_call_id: None,
            name: None,
            metadata: Default::default(),
        },
        None => ChatMessage::user(text),
    };
    if let Some(ct) = v.get("customType").and_then(Value::as_str) {
        msg.metadata
            .insert("pi_custom_type".to_string(), ct.to_string());
    }
    if let Some(d) = v.get("display").and_then(Value::as_bool) {
        msg.metadata.insert("pi_display".to_string(), d.to_string());
    }
    if let Some(details) = v.get("details") {
        if !details.is_null() {
            msg.metadata
                .insert("pi_details".to_string(), details.to_string());
        }
    }
    // WAVE-2 fidelity fix: preserve pi's message-level unix-ms clock
    // separately from the canonical entry-level ISO `timestamp` — see
    // `push_pi_user`. `v` here is the `message` object for the `role:
    // "custom"` case; for the top-level `custom_message` case `v` is the
    // entry itself, whose `timestamp` is the entry-level ISO string (not a
    // u64), so this is a no-op there — matching pre-WAVE-2 behavior.
    if let Some(ts) = v.get("timestamp").and_then(Value::as_u64) {
        msg.metadata
            .insert("pi_msg_timestamp".to_string(), ts.to_string());
    }
    out.push(msg);
}

/// pi's own prefix-wrapped user text for a `compaction` entry summary
/// (§2.1: "compaction/branch summaries (pi's own prefix-wrapped user text)").
/// The exact upstream wrapper string is cited (`msg:11-17`) but not
/// reproduced in the frozen research doc; this is a clearly-labeled
/// reconstruction, not pi's literal bytes — see build-report ambiguity note.
fn push_pi_compaction(entry_v: &Value, out: &mut Vec<ChatMessage>) {
    let summary = entry_v.get("summary").and_then(Value::as_str).unwrap_or("");
    if summary.trim().is_empty() {
        return;
    }
    let mut msg = ChatMessage::user(format!("[compaction summary]\n{summary}"));
    msg.metadata
        .insert("pi_type".to_string(), "compaction".to_string());
    if let Some(fk) = entry_v.get("firstKeptEntryId").and_then(Value::as_str) {
        msg.metadata
            .insert("pi_first_kept_entry_id".to_string(), fk.to_string());
    }
    if let Some(tb) = entry_v.get("tokensBefore").and_then(Value::as_u64) {
        msg.metadata
            .insert("pi_tokens_before".to_string(), tb.to_string());
    }
    if let Some(d) = entry_v.get("details") {
        if !d.is_null() {
            msg.metadata.insert("pi_details".to_string(), d.to_string());
        }
    }
    if entry_v.get("fromHook").and_then(Value::as_bool) == Some(true) {
        msg.metadata
            .insert("pi_from_hook".to_string(), "true".to_string());
    }
    out.push(msg);
}

/// pi's own prefix-wrapped user text for a `branch_summary` entry (a
/// rewind-with-summary) — same reconstruction caveat as
/// [`push_pi_compaction`].
fn push_pi_branch_summary(entry_v: &Value, out: &mut Vec<ChatMessage>) {
    let summary = entry_v.get("summary").and_then(Value::as_str).unwrap_or("");
    if summary.trim().is_empty() {
        return;
    }
    let mut msg = ChatMessage::user(format!("[branch summary]\n{summary}"));
    msg.metadata
        .insert("pi_type".to_string(), "branch_summary".to_string());
    if let Some(f) = entry_v.get("fromId").and_then(Value::as_str) {
        msg.metadata.insert("pi_from_id".to_string(), f.to_string());
    }
    if let Some(d) = entry_v.get("details") {
        if !d.is_null() {
            msg.metadata.insert("pi_details".to_string(), d.to_string());
        }
    }
    if entry_v.get("fromHook").and_then(Value::as_bool) == Some(true) {
        msg.metadata
            .insert("pi_from_hook".to_string(), "true".to_string());
    }
    out.push(msg);
}

/// pi's own message-level unix-ms clock (`metadata["pi_msg_timestamp"]`,
/// restored by the pi loader's `push_pi_*` helpers) — DISTINCT from the
/// canonical entry-level ISO `metadata["timestamp"]` [`msg_timestamp_or_synth`]
/// reads. The two carry genuinely different values in real pi corpora (a
/// message-level clock reading vs. the entry's own wall-clock stamp), so this
/// is intentionally its own accessor. Used ONLY by [`Session::write_pi_entries`]'s
/// nested `message.timestamp` field, so a pi -> pi native round-trip
/// preserves the source message-level clock value-exact instead of deriving
/// it from the (distinct) entry-level timestamp. Falls back to
/// [`SYNTH_TS_MS`] for a message that never carried a pi message-level clock
/// reading (non-pi-sourced, or a synthesized/appended turn).
fn msg_pi_native_timestamp_ms(msg: &ChatMessage) -> i64 {
    msg.metadata
        .get("pi_msg_timestamp")
        .and_then(|s| s.parse::<i64>().ok())
        .unwrap_or(SYNTH_TS_MS)
}

impl Session {
    /// Synthesize a fresh pi v3 session from the canonical `messages`
    /// (T3 cross-format/full synthesis — `to_jsonl(other)`'s "view round-trip"
    /// tier, §3/§4.1: this is NOT the byte-lossless native path, which goes
    /// through `raw` + `to_native_jsonl(_v2)` instead).
    pub(super) fn to_pi_jsonl(&self) -> String {
        let session_id = self
            .meta
            .session_id
            .clone()
            .unwrap_or_else(|| synth_uuid(0));
        let cwd = self.cwd_string();
        let mut out = String::new();
        push_pi_header(
            &mut out,
            &session_id,
            &cwd,
            self.meta
                .lineage
                .get("parent_session_path")
                .map(String::as_str),
            self.meta.lineage.get("created_at").map(String::as_str),
            // D7: carry a captured Claude `fork-context-ref` (see
            // `capture_claude_meta`) through the Pi hop too — mirrors the
            // Codex hop's `claude_fork_context_ref` passthrough
            // (`write_synthesized_codex_header`) so a Claude -> Pi -> Claude
            // round trip doesn't silently lose fork lineage just because Pi
            // has no native slot for it.
            self.meta
                .lineage
                .get("claude_fork_context_ref_raw")
                .map(String::as_str),
        );
        let mut used_ids: HashSet<String> = HashSet::new();
        let mut counter: u64 = 0;
        self.write_pi_entries(&mut out, &self.messages, None, &mut used_ids, &mut counter);
        if let Some(extension) = native_residue_envelope(&self.meta) {
            inject_first_jsonl_top_level(
                &mut out,
                SUPERCODE_NATIVE_RESIDUE_SUMMARY_KEY,
                native_residue_summary(&extension),
            );
            inject_first_jsonl_top_level(&mut out, SUPERCODE_NATIVE_RESIDUE_KEY, extension);
        }
        out
    }

    /// Synthesize pi `message` entries for `messages` (a full session, or —
    /// for [`Self::to_pi_jsonl_spliced`] — just the appended tail), chaining
    /// `parentId` from `parent` and drawing fresh ids from `used_ids`/
    /// `counter`. Skips [`is_replay_excluded`] messages (`compacted_out`,
    /// `pi_exclude_from_context`) exactly like the Claude/Codex writers.
    fn write_pi_entries(
        &self,
        out: &mut String,
        messages: &[ChatMessage],
        mut parent: Option<String>,
        used_ids: &mut HashSet<String>,
        counter: &mut u64,
    ) {
        // Claude Code and Codex do not repeat the tool name on their native
        // tool-result records. Recover that redundant Pi field from the
        // paired assistant call when a cross-format round trip therefore
        // returns a canonical Tool message with `name == None`.
        let mut paired_tool_names = HashMap::<String, String>::new();
        for msg in messages {
            if is_replay_excluded(msg) {
                continue;
            }
            for call in msg.tool_calls() {
                paired_tool_names.insert(call.id.clone(), call.function.name.clone());
            }
            let id = pi_fresh_id(used_ids, counter);
            let mut entry = match msg.role {
                // B4: pi has no session-level system/developer PROMPT slot
                // (§1.1: "no system-prompt... rebuilt at runtime"), but a
                // content-bearing `Role::System` message loaded from a real
                // Claude Code `type: "system"` record (`push_claude_system`'s
                // keep-listed subtypes: `local_command`, `scheduled_task_fire`,
                // `away_summary`) is NOT a system prompt — it's a real,
                // non-regenerable transcript event. Pi's own `role:"custom"`
                // `CustomMessage` (§3e: "extension-injected... sent to the LLM
                // as a user message") is the closest existing, non-fabricated
                // slot pi's own parser already understands, so this
                // re-materializes the record there instead of silently
                // dropping it — the exact allowance push_claude_system's own
                // doc comment describes in reverse. `customType` is a
                // supercode-namespaced marker (`push_pi_custom_common`
                // recognizes it on reload and restores `Role::System` +
                // `metadata["systemSubtype"]`, exactly like `push_claude_system`
                // produced in the first place); a real pi customType never
                // collides with this name. `details.claude_system_subtype`
                // carries the original subtype losslessly through the pi leg
                // (mirrors `write_codex_records`'s `claude_system_subtype`
                // metadata channel on the Codex leg, PARITY-6 dev/02). Content
                // is never fabricated — only emitted when non-empty.
                Role::System => {
                    let content = msg.content.clone().unwrap_or_default();
                    if content.trim().is_empty() {
                        continue;
                    }
                    let subtype = msg
                        .metadata
                        .get("systemSubtype")
                        .cloned()
                        .unwrap_or_else(|| "local_command".to_string());
                    serde_json::json!({
                        "type": "message",
                        "id": id,
                        "parentId": parent,
                        "timestamp": msg_timestamp_or_synth(msg),
                        "message": {
                            "role": "custom",
                            "customType": PI_CLAUDE_SYSTEM_CUSTOM_TYPE,
                            "content": content,
                            "display": true,
                            "details": {"claude_system_subtype": subtype},
                            "timestamp": msg_pi_native_timestamp_ms(msg),
                        },
                    })
                }
                Role::User => serde_json::json!({
                    "type": "message",
                    "id": id,
                    "parentId": parent,
                    "timestamp": msg_timestamp_or_synth(msg),
                    "message": {
                        "role": "user",
                        "content": pi_content_value(msg),
                        "timestamp": msg_pi_native_timestamp_ms(msg),
                    },
                }),
                Role::Assistant => {
                    let api = msg
                        .metadata
                        .get("pi_api")
                        .cloned()
                        .unwrap_or_else(|| "anthropic-messages".to_string());
                    let provider = msg
                        .metadata
                        .get("pi_provider")
                        .cloned()
                        .unwrap_or_else(|| "anthropic".to_string());
                    let model = self
                        .meta
                        .model
                        .clone()
                        .unwrap_or_else(|| "unknown".to_string());
                    let usage = msg
                        .metadata
                        .get("pi_usage")
                        .and_then(|s| serde_json::from_str::<Value>(s).ok())
                        .unwrap_or_else(default_pi_usage);
                    let stop_reason = msg
                        .metadata
                        .get("pi_stop_reason")
                        .cloned()
                        .unwrap_or_else(|| "stop".to_string());
                    serde_json::json!({
                        "type": "message",
                        "id": id,
                        "parentId": parent,
                        "timestamp": msg_timestamp_or_synth(msg),
                        "message": {
                            "role": "assistant",
                            "content": pi_assistant_content_value(msg),
                            "api": api,
                            "provider": provider,
                            "model": model,
                            "usage": usage,
                            "stopReason": stop_reason,
                            "timestamp": msg_pi_native_timestamp_ms(msg),
                        },
                    })
                }
                Role::Tool => serde_json::json!({
                    "type": "message",
                    "id": id,
                    "parentId": parent,
                    "timestamp": msg_timestamp_or_synth(msg),
                    "message": {
                        "role": "toolResult",
                        "toolCallId": msg.tool_call_id.clone().unwrap_or_default(),
                        "toolName": msg.name.as_deref().or_else(|| {
                            msg.tool_call_id
                                .as_deref()
                                .and_then(|id| paired_tool_names.get(id).map(String::as_str))
                        }).unwrap_or_default(),
                        "content": pi_tool_result_content(msg),
                        "isError": is_tool_error_flag(msg),
                        "timestamp": msg_pi_native_timestamp_ms(msg),
                    },
                }),
            };
            if msg.role == Role::Tool && crate::tool_outcome(msg) == crate::ToolOutcome::Unknown {
                entry[SUPERCODE_TOOL_OUTCOME_KEY] = Value::String("unknown".to_string());
            }
            set_grok_message_extension(&mut entry, self.meta.source, msg);
            push_jsonl(out, &entry);
            parent = Some(id);
            if msg.role == Role::Tool {
                if let Some(call_id) = msg.tool_call_id.as_deref() {
                    paired_tool_names.remove(call_id);
                }
            }
        }
    }

    /// A12-style splice for pi (§1.1/§4.2): replay the imported `raw` prefix
    /// **verbatim** — the header line always has its `version` normalized to
    /// 3 (§1.3: pi rewrites any pre-v3 file in place on load, destroying
    /// byte-identity, so the writer never re-emits one; this intentionally
    /// breaks byte-identity for pre-v3 originals only, the accepted
    /// trade-off) and its `id` overridden when `session_id` is `Some`. Every
    /// other raw line — every entry — is untouched (pi repeats the session
    /// id on no other line, `pi-fields.md` §1). Then synthesizes `message`
    /// entries only for the appended tail via [`Self::write_pi_entries`],
    /// chaining from the last entry `id` found in the raw prefix.
    pub(super) fn to_pi_jsonl_spliced(&self, session_id: Option<&str>) -> Result<String> {
        let (raw_prefix_len, message_prefix_len) = self.spliced_prefix_lens();
        if raw_prefix_len == 0 {
            return Ok(self.to_pi_jsonl());
        }

        let mut out = String::new();
        let mut used_ids: HashSet<String> = HashSet::new();
        let mut leaf: Option<String> = None;
        for (i, line) in self.raw[..raw_prefix_len].iter().enumerate() {
            if i == 0 {
                if let Ok(v) = serde_json::from_str::<Value>(line) {
                    if v.get("type").and_then(Value::as_str) == Some("session") {
                        let needs_v3 = v.get("version").and_then(Value::as_u64) != Some(3);
                        // Only reparse+reserialize the header when something
                        // actually needs to change — this crate doesn't
                        // enable serde_json's `preserve_order`, so a no-op
                        // round-trip through `Value` would reorder keys
                        // alphabetically and silently break the "prefix
                        // bytes unchanged" splice guarantee for the (common)
                        // already-v3, no-override case.
                        if needs_v3 || session_id.is_some() {
                            let mut v = v;
                            v["version"] = serde_json::json!(3);
                            if let Some(new_id) = session_id {
                                v["id"] = Value::String(new_id.to_string());
                            }
                            out.push_str(&v.to_string());
                            out.push('\n');
                            continue;
                        }
                    }
                }
            }
            out.push_str(line);
            out.push('\n');
            if let Ok(v) = serde_json::from_str::<Value>(line) {
                if let Some(id) = v.get("id").and_then(Value::as_str) {
                    used_ids.insert(id.to_string());
                    leaf = Some(id.to_string());
                }
            }
        }

        let mut counter: u64 = 0;
        self.write_pi_entries(
            &mut out,
            &self.messages[message_prefix_len..],
            leaf,
            &mut used_ids,
            &mut counter,
        );
        Ok(out)
    }
}

// ---- Pi writer helpers -----------------------------------------------------

/// Emit the pi v3 `session` header line. Frozen (§1.1/§1.3): `version` is
/// ALWAYS 3, never a lower value, so pi never rewrites a supercode-emitted
/// file in place on first resume (`pi-fields.md` sm:848-850).
fn push_pi_header(
    out: &mut String,
    id: &str,
    cwd: &str,
    parent_session: Option<&str>,
    created_at: Option<&str>,
    claude_fork_context_ref: Option<&str>,
) {
    let mut header = serde_json::json!({
        "type": "session",
        "version": 3,
        "id": id,
        "timestamp": created_at.unwrap_or(SYNTH_TS),
        "cwd": cwd,
    });
    if let Some(ps) = parent_session {
        header["parentSession"] = Value::String(ps.to_string());
    }
    // D7: namespaced passthrough field, exactly like the Codex writer's
    // `session_meta.payload.claude_fork_context_ref` — pi tolerates unknown
    // header keys, and `capture_pi_header` reads this same key back on
    // import, so a Claude -> Pi -> Claude round trip still reconstructs the
    // fork-context-ref record instead of silently losing it on this hop.
    if let Some(raw) = claude_fork_context_ref {
        header["claude_fork_context_ref"] =
            serde_json::from_str(raw).unwrap_or_else(|_| Value::String(raw.to_string()));
    }
    push_jsonl(out, &header);
}

/// A fresh 8-hex entry id, collision-checked against `used` (mirrors pi's own
/// `randomUUID().slice(0,8)` + collision check, `pi-fields.md` sm:216-224 —
/// deterministic here rather than random, which still satisfies "fresh,
/// collision-free" without an extra RNG dependency).
fn pi_fresh_id(used: &mut HashSet<String>, counter: &mut u64) -> String {
    loop {
        *counter += 1;
        let h = counter.wrapping_mul(0x9E3779B97F4A7C15);
        let id = format!("{:08x}", (h >> 32) as u32);
        if used.insert(id.clone()) {
            return id;
        }
    }
}

/// Parse a `data:<mime>[;base64],<data>` URI back into `(mime, data)` — the
/// inverse of the loader's `data:{mime};base64,{data}` construction.
pub(super) fn parse_data_uri(url: &str) -> Option<(String, String)> {
    let rest = url.strip_prefix("data:")?;
    let (meta, data) = rest.split_once(',')?;
    let mime = meta.strip_suffix(";base64").unwrap_or(meta);
    Some((mime.to_string(), data.to_string()))
}

/// Rebuild a pi `(TextContent|ImageContent)[]` (or bare string) content value
/// from a `ChatMessage`'s `content`/`content_parts` — shared by `user` and
/// `toolResult` entries (both use the identical union on the wire).
fn pi_content_value(msg: &ChatMessage) -> Value {
    if let Some(parts) = &msg.content_parts {
        let mut arr = 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) {
                        arr.push(serde_json::json!({"type": "text", "text": t}));
                    }
                }
                Some("image_url") => {
                    if let Some(url) = p
                        .get("image_url")
                        .and_then(|u| u.get("url"))
                        .and_then(Value::as_str)
                    {
                        if let Some((mime, data)) = parse_data_uri(url) {
                            arr.push(
                                serde_json::json!({"type": "image", "mimeType": mime, "data": data}),
                            );
                        }
                    }
                }
                _ => {}
            }
        }
        Value::Array(arr)
    } else {
        Value::String(msg.content.clone().unwrap_or_default())
    }
}

/// A Pi `toolResult`'s content is always a block list (a user message may be a bare string):
/// stock Pi reads a string here character by character and the resumed model sees blank output.
fn pi_tool_result_content(msg: &ChatMessage) -> Value {
    match pi_content_value(msg) {
        Value::String(text) if text.is_empty() => Value::Array(Vec::new()),
        Value::String(text) => serde_json::json!([{"type": "text", "text": text}]),
        blocks => blocks,
    }
}

fn pi_assistant_content_value(msg: &ChatMessage) -> Value {
    let mut arr = Vec::new();
    if let Some(thinking) = msg.metadata.get("thinking") {
        let mut block = serde_json::json!({"type": "thinking", "thinking": thinking});
        if let Some(sig) = msg.metadata.get("thinking_signature") {
            block["thinkingSignature"] = Value::String(sig.clone());
        }
        if msg.metadata.get("pi_thinking_redacted").map(String::as_str) == Some("true") {
            block["redacted"] = Value::Bool(true);
        }
        arr.push(block);
    }
    if let Some(text) = &msg.content {
        if !text.is_empty() {
            let mut block = serde_json::json!({"type": "text", "text": text});
            if let Some(sig) = msg.metadata.get("pi_text_signature") {
                block["textSignature"] = Value::String(sig.clone());
            }
            arr.push(block);
        }
    }
    for tc in msg.tool_calls() {
        let args = tc
            .function
            .parsed_arguments()
            .unwrap_or_else(|_| Value::Object(Default::default()));
        let mut block = serde_json::json!({
            "type": "toolCall",
            "id": tc.id,
            "name": tc.function.name,
            "arguments": args,
        });
        if let Some(sig) = msg.metadata.get("pi_thought_signature") {
            block["thoughtSignature"] = Value::String(sig.clone());
        }
        arr.push(block);
    }
    Value::Array(arr)
}

fn default_pi_usage() -> Value {
    serde_json::json!({
        "input": 0, "output": 0, "cacheRead": 0, "cacheWrite": 0, "totalTokens": 0,
        "cost": {"input": 0.0, "output": 0.0, "cacheRead": 0.0, "cacheWrite": 0.0, "total": 0.0},
    })
}

fn is_tool_error_flag(msg: &ChatMessage) -> bool {
    msg.metadata.get("pi_is_error").map(String::as_str) == Some("true") || crate::is_tool_error(msg)
}