supercode-harness 0.4.8

//! Wave A conformance tests for the Pi loader/writer
//! (`docs/interop/opencode-pi-spec.md` §1.1/§4.1/§6, `pi-fields.md`).
//!
//! Mirrors the existing `roundtrip_regression.rs`/`schema_coverage.rs`
//! patterns: everything here runs under a plain `cargo test`, against the
//! committed `pi_session.jsonl` fixture, no env gate, no `#[ignore]`.

mod interop_common;

use std::path::{Path, PathBuf};

use interop_common::msg_eq_multimodal;
use supercode_harness::audit::{audit_dir, Corpus};
use supercode_harness::session::{Session, SessionFormat};
use supercode_harness::{ChatMessage, Role};

fn fixture(name: &str) -> PathBuf {
    Path::new(env!("CARGO_MANIFEST_DIR"))
        .join("tests/fixtures")
        .join(name)
}

// ---- (a) the real "100%": byte-lossless native round-trip ------------------

/// Strip the leading `to_native_jsonl(_v2)` envelope header line (the
/// `{"supercode_native": N, "source": "pi", ...}` line that the native
/// wrapper prepends) and return the remaining body — the part that is
/// supposed to be the source file's own bytes, verbatim, one raw line per
/// line, each `\n`-terminated (`to_native_jsonl`'s own construction: `raw`
/// lines pushed with a trailing `\n` each, and the fixture itself is
/// well-formed JSONL that already ends in `\n`). This direct strip-header
/// comparison only works for a source that both ends in `\n` and has no
/// interior blank lines/CRLF — for arbitrary/pathological input use
/// `join_source` (below `pi_raw_capture_is_strict_verbatim_for_pathological_input`),
/// which reconstructs via `raw` + `raw_trailing_newline` (IX-1) and handles
/// any input.
fn strip_native_header(native: &str) -> &str {
    let idx = native
        .find('\n')
        .expect("native output must have a header line");
    &native[idx + 1..]
}

/// The genuine, non-circular byte-lossless proof: read the fixture as raw
/// bytes straight off disk, run it through `from_pi_str` -> `to_native_jsonl`
/// (and the `_v2` sibling), and diff the RECONSTRUCTED bytes against the
/// ORIGINAL FILE BYTES — not against another derivation of `raw` computed
/// with the same `non_empty_lines` transform the loader itself uses (that
/// older shape only proved the transform is idempotent/deterministic, not
/// that it preserves the source file's bytes).
#[test]
fn pi_native_round_trip_is_byte_lossless() {
    let path = fixture("pi_session.jsonl");
    let original_bytes = std::fs::read(&path).unwrap();
    let original_text = std::str::from_utf8(&original_bytes).expect("fixture must be valid UTF-8");

    let session = Session::from_pi(&path).unwrap();

    // to_native_jsonl -> strip its own envelope header -> must equal the
    // ORIGINAL FILE BYTES exactly. This is a real source-file-vs-
    // reconstructed-output comparison, not raw-vs-raw.
    let native = session.to_native_jsonl();
    let reconstructed_body = strip_native_header(&native);
    assert_eq!(
        reconstructed_body.as_bytes(),
        original_bytes.as_slice(),
        "pi native round-trip must reproduce the ORIGINAL FIXTURE FILE'S BYTES exactly \
         (T1-byte, genuine source-vs-reconstructed diff, not raw-vs-raw)"
    );
    assert_eq!(
        reconstructed_body, original_text,
        "sanity: byte comparison and str comparison must agree"
    );

    // v2 (appended-turn machinery) with zero appended turns must be equally
    // lossless against the ORIGINAL FILE BYTES — this is the path the
    // emulate-to-continue proof (§4.2) builds on. Its header differs from v1
    // (carries `session_id`/`created`), but the body after that header must
    // still equal the original file byte-for-byte.
    let native_v2 = session.to_native_jsonl_v2(&[]);
    let reconstructed_body_v2 = strip_native_header(&native_v2);
    assert_eq!(
        reconstructed_body_v2.as_bytes(),
        original_bytes.as_slice(),
        "pi native-v2 round-trip (no appended turns) must reproduce the ORIGINAL FIXTURE \
         FILE'S BYTES exactly"
    );

    // And the full from_native_str -> raw reload agrees with a fresh
    // from_pi_str parse of the original file — the round-trip is coherent at
    // the `Session` level too, not just at the raw-body-bytes level.
    let reloaded = Session::from_native_str(&native).unwrap();
    assert_eq!(session.raw, reloaded.raw);
    let reloaded_v2 = Session::from_native_str(&native_v2).unwrap();
    assert_eq!(session.raw, reloaded_v2.raw);

    println!(
        "pi T1-byte native round-trip: {} bytes byte-equal against the ORIGINAL FIXTURE \
         FILE (source-vs-reconstructed) — PASS (the real \"100%\")",
        original_bytes.len()
    );
}

/// Reconstruct the ORIGINAL source bytes `session.raw` was captured from:
/// the exact inverse of the crate-internal `split_lines_verbatim` (IX-1) —
/// join `raw`'s lines with `\n`, then append one more `\n` iff the source
/// had a trailing newline. `raw`'s line list alone can't tell a
/// trailing-newline source from one that doesn't have one (both split into
/// the same lines), which is exactly why `raw_trailing_newline` exists.
fn join_source(session: &Session) -> String {
    let mut out = session.raw.join("\n");
    if session.raw_trailing_newline {
        out.push('\n');
    }
    out
}

/// IX-1: strict-verbatim raw capture. Supersedes the old
/// `pi_raw_capture_normalizes_blank_and_crlf_lines` (deleted), which
/// documented `non_empty_lines`-based capture NORMALIZING blank lines / CRLF
/// / trailing whitespace away — i.e. the native round-trip was only
/// byte-lossless for already-well-formed input. `Session.raw` is now
/// captured via `split_lines_verbatim`, a separate strict-verbatim path from
/// the blank-skipping/trimming `non_empty_lines` PARSE walk (which keeps
/// skipping blank lines when it looks for records — see
/// `pi_parsing_skips_blank_lines_between_records` below) — so this exact
/// pathological input now round-trips byte-for-byte.
#[test]
fn pi_raw_capture_is_strict_verbatim_for_pathological_input() {
    let header = r#"{"type":"session","version":3,"id":"s1","timestamp":"2026-01-01T00:00:00.000Z","cwd":"/tmp"}"#;
    let msg = r#"{"type":"message","id":"a1","parentId":null,"timestamp":"2026-01-01T00:00:01.000Z","message":{"role":"user","content":"hi","timestamp":1}}"#;
    let msg2 = r#"{"type":"message","id":"a2","parentId":"a1","timestamp":"2026-01-01T00:00:02.000Z","message":{"role":"assistant","content":"hi back","timestamp":2}}"#;

    // Blank lines between/around records, a trailing-whitespace-padded
    // header, a CRLF line ending on the first message, AND no trailing
    // newline at EOF (the very last byte is the closing `}` of `msg2`).
    let pathological = format!("{header}   \n\n\n{msg}\r\n\n{msg2}");
    assert!(
        !pathological.ends_with('\n'),
        "sanity: this input has no trailing newline at EOF"
    );

    let session = Session::from_pi_str(&pathological).unwrap();

    // load -> to_native_jsonl -> from_native_str -> reconstructed source
    // bytes must equal the ORIGINAL pathological bytes exactly.
    let native = session.to_native_jsonl();
    let reloaded = Session::from_native_str(&native).unwrap();
    assert_eq!(
        join_source(&reloaded).as_bytes(),
        pathological.as_bytes(),
        "IX-1: strict-verbatim raw capture must round-trip pathological pi JSONL (blank \
         lines / CRLF / trailing whitespace / no trailing newline at EOF) byte-exact \
         through load -> to_native_jsonl -> from_native_str"
    );

    // The v2/sidecar native path (zero appended turns) must be equally
    // byte-exact — this is the path the reduction engine's sidecar round
    // trip builds on.
    let native_v2 = session.to_native_jsonl_v2(&[]);
    let reloaded_v2 = Session::from_native_str(&native_v2).unwrap();
    assert_eq!(
        join_source(&reloaded_v2).as_bytes(),
        pathological.as_bytes(),
        "IX-1: v2/sidecar native round-trip must also be byte-exact for pathological input"
    );
}

/// IX-1 companion (dev/02): a blank line between two records must still be
/// SKIPPED by parsing — never a parse error, never a spurious empty message
/// — even though `raw` now preserves it verbatim (the test above). A blank
/// line is simply not a JSON record on either view.
#[test]
fn pi_parsing_skips_blank_lines_between_records() {
    let header = r#"{"type":"session","version":3,"id":"s1","timestamp":"2026-01-01T00:00:00.000Z","cwd":"/tmp"}"#;
    let msg1 = r#"{"type":"message","id":"a1","parentId":null,"timestamp":"2026-01-01T00:00:01.000Z","message":{"role":"user","content":"hi","timestamp":1}}"#;
    let msg2 = r#"{"type":"message","id":"a2","parentId":"a1","timestamp":"2026-01-01T00:00:02.000Z","message":{"role":"assistant","content":[{"type":"text","text":"hi back"}],"timestamp":2}}"#;
    let jsonl = format!("{header}\n{msg1}\n\n{msg2}\n");

    let session = Session::from_pi_str(&jsonl).unwrap();

    assert_eq!(
        session.messages.len(),
        2,
        "the blank line between the two records must not produce a spurious empty \
         message or a parse error: {:?}",
        session.messages
    );
    assert_eq!(session.messages[0].content.as_deref(), Some("hi"));
    assert_eq!(session.messages[1].content.as_deref(), Some("hi back"));

    // But `raw` (strict-verbatim, IX-1) DOES retain the blank line.
    assert_eq!(
        session.raw.len(),
        4,
        "raw must retain all 4 lines verbatim, including the blank one"
    );
    assert_eq!(
        session.raw[2], "",
        "the blank line itself must survive in raw"
    );
}

/// Only Pi's truly contentless `stopReason:error` record is raw-only
/// residue. An empty aborted turn is still transcript state, and a
/// thinking-bearing error turn has native replayable content; neither may
/// be swept up by the empty-error exception.
#[test]
fn pi_empty_error_exception_does_not_drop_aborted_or_thinking_turns() {
    let jsonl = [
        r#"{"type":"session","version":3,"id":"s-empty-boundary","timestamp":"2026-01-01T00:00:00.000Z","cwd":"/tmp"}"#,
        r#"{"type":"message","id":"m-error","parentId":null,"timestamp":"2026-01-01T00:00:01.000Z","message":{"role":"assistant","content":[],"stopReason":"error","errorMessage":"overloaded","timestamp":1}}"#,
        r#"{"type":"message","id":"m-aborted","parentId":"m-error","timestamp":"2026-01-01T00:00:02.000Z","message":{"role":"assistant","content":[],"stopReason":"aborted","errorMessage":"Request was aborted","timestamp":2}}"#,
        r#"{"type":"message","id":"m-thinking-error","parentId":"m-aborted","timestamp":"2026-01-01T00:00:03.000Z","message":{"role":"assistant","content":[{"type":"thinking","thinking":"retry analysis"}],"stopReason":"error","errorMessage":"failed after thinking","timestamp":3}}"#,
    ]
    .join("\n");

    let session = Session::from_pi_str(&jsonl).expect("boundary fixture must load");
    assert_eq!(
        session.messages.len(),
        2,
        "only the truly contentless error record should stay raw-only"
    );
    let aborted = &session.messages[0];
    assert_eq!(
        aborted.metadata.get("pi_stop_reason").map(String::as_str),
        Some("aborted")
    );
    assert_eq!(
        aborted
            .metadata
            .get("empty_assistant_record")
            .map(String::as_str),
        Some("true")
    );
    let thinking_error = &session.messages[1];
    assert_eq!(
        thinking_error.metadata.get("thinking").map(String::as_str),
        Some("retry analysis")
    );
    assert_eq!(
        thinking_error
            .metadata
            .get("pi_stop_reason")
            .map(String::as_str),
        Some("error")
    );
}

// ---- (b) canonical `messages` correctness ----------------------------------

#[test]
fn pi_canonical_messages_map_correctly() {
    let session = Session::from_pi(fixture("pi_session.jsonl")).unwrap();

    // user -> Role::User, plain text.
    assert!(
        session.messages.iter().any(|m| m.role == Role::User
            && m.content.as_deref() == Some("Please create hello.txt with 'hi' in it.")),
        "plain user text message must map to Role::User"
    );

    // assistant text + toolCall -> Role::Assistant with tool_calls, arguments
    // round-tripping through the object->string serialization value-lossless.
    let assistant_with_call = session
        .messages
        .iter()
        .find(|m| m.role == Role::Assistant && !m.tool_calls().is_empty())
        .expect("an assistant message with a tool call");
    let call = &assistant_with_call.tool_calls()[0];
    assert_eq!(call.function.name, "write_file");
    assert_eq!(
        call.function.parsed_arguments().unwrap(),
        serde_json::json!({"path": "hello.txt", "content": "hi"})
    );

    // toolResult -> Role::Tool, paired by id, name carried.
    let tool_result = session
        .messages
        .iter()
        .find(|m| m.role == Role::Tool)
        .expect("a tool result message");
    assert_eq!(tool_result.tool_call_id.as_deref(), Some(call.id.as_str()));
    assert_eq!(tool_result.name.as_deref(), Some("write_file"));
    assert_eq!(
        tool_result.content.as_deref(),
        Some("wrote 2 bytes to hello.txt")
    );

    // bashExecution -> rendered as Role::User text, structured fields kept
    // in metadata for fidelity.
    let bash = session
        .messages
        .iter()
        .find(|m| m.metadata.contains_key("pi_bash_command"))
        .expect("a bashExecution message");
    assert_eq!(bash.role, Role::User);
    assert_eq!(
        bash.metadata.get("pi_bash_command").map(String::as_str),
        Some("cat hello.txt")
    );
    assert_eq!(
        bash.metadata.get("pi_bash_output").map(String::as_str),
        Some("hi")
    );
    assert!(bash
        .content
        .as_deref()
        .unwrap_or("")
        .contains("cat hello.txt"));

    // custom_message -> Role::User, customType/display preserved.
    let custom = session
        .messages
        .iter()
        .find(|m| m.metadata.get("pi_custom_type").map(String::as_str) == Some("note"))
        .expect("the custom_message entry");
    assert_eq!(custom.role, Role::User);
    assert_eq!(custom.content.as_deref(), Some("reminder: keep going"));
    assert_eq!(
        custom.metadata.get("pi_display").map(String::as_str),
        Some("true")
    );

    // multimodal: image content_part decodes to the exact fixture bytes
    // (§4.1 S2 floor — the fixture must exercise this path).
    let img_msg = session
        .messages
        .iter()
        .find(|m| m.content_parts.is_some())
        .expect("a multimodal user message");
    let parts = img_msg.content_parts.as_ref().unwrap();
    assert!(
        parts
            .iter()
            .any(|p| p.get("type").and_then(|v| v.as_str()) == Some("text")),
        "multimodal message must keep its leading text part"
    );
    let image_part = parts
        .iter()
        .find(|p| p.get("type").and_then(|v| v.as_str()) == Some("image_url"))
        .expect("an image_url part");
    let url = image_part["image_url"]["url"].as_str().unwrap();
    let expected_b64 =
        "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mNk+A8AAQUBAScY42YAAAAASUVORK5CYII=";
    assert_eq!(
        url,
        format!("data:image/png;base64,{expected_b64}"),
        "image_url data URI must carry pi's exact mimeType+data bytes"
    );

    // compaction discipline (§2.1/§2.2 S3): pre-compaction messages carry
    // compacted_out="true"; the kept tail and the summary itself do not.
    let compacted: Vec<&ChatMessage> = session
        .messages
        .iter()
        .filter(|m| m.metadata.get("compacted_out").map(String::as_str) == Some("true"))
        .collect();
    assert!(
        !compacted.is_empty(),
        "the fixture's compaction boundary must mark at least one message compacted_out"
    );
    assert!(
        compacted
            .iter()
            .any(|m| m.content.as_deref() == Some("Done! Created hello.txt.")),
        "the pre-compaction assistant turn must be marked compacted_out"
    );
    assert!(
        session
            .messages
            .iter()
            .any(|m| m.metadata.contains_key("pi_bash_command")
                && !m.metadata.contains_key("compacted_out")),
        "the kept-tail start (firstKeptEntryId) must NOT be marked compacted_out"
    );
    let summary_msg = session
        .messages
        .iter()
        .find(|m| m.metadata.get("pi_type").map(String::as_str) == Some("compaction"))
        .expect("the compaction summary message");
    assert!(
        !summary_msg.metadata.contains_key("compacted_out"),
        "the compaction summary itself must never be marked compacted_out"
    );
    assert_eq!(
        summary_msg
            .metadata
            .get("pi_first_kept_entry_id")
            .map(String::as_str),
        Some("e6000006")
    );

    // §5.1(f) wants BOTH a plain abandoned branch AND a rewind-with-summary
    // that sits ON the active path. The fixture now has both:
    //   - `e2b00002`/`ebranch1`: a plain abandoned branch off `e1000001`
    //     whose `branch_summary` is itself off the active path too — neither
    //     enters `messages` at all (raw-only residue).
    //   - `e4b00004`/`ebranchAct1`: a second abandoned sibling off
    //     `e4000004`, but its `branch_summary` (`ebranchAct1`) is ON the
    //     active path (`e4000004 -> ebranchAct1 -> e5000005 -> ...`), so
    //     `push_pi_branch_summary` fires for real here (FIX #3).
    assert!(
        !session.messages.iter().any(|m| m
            .content
            .as_deref()
            .unwrap_or("")
            .contains("later abandoned")),
        "the plain abandoned branch's assistant turn must not appear in canonical messages"
    );
    assert!(
        session.raw.iter().any(|l| l.contains("later abandoned")),
        "the plain abandoned branch must still survive verbatim in raw"
    );
    assert!(
        !session.messages.iter().any(|m| m
            .content
            .as_deref()
            .unwrap_or("")
            .contains("discarded before sending")),
        "the active-path branch_summary's OWN abandoned sibling (e4b00004) must not \
         appear in canonical messages — only the branch_summary itself does"
    );
    assert!(
        session
            .raw
            .iter()
            .any(|l| l.contains("\"type\":\"branch_summary\"")),
        "a branch_summary record must survive verbatim in raw"
    );
    assert_eq!(
        session
            .raw
            .iter()
            .filter(|l| l.contains("\"type\":\"branch_summary\""))
            .count(),
        2,
        "the fixture must carry both the off-active and the active-path branch_summary"
    );

    // FIX #3: the ACTIVE-path branch_summary (`ebranchAct1`) maps through
    // `push_pi_branch_summary` to a canonical `Role::User` message, exactly
    // like `push_pi_compaction` does for `compaction` — this is the
    // previously-dead code path, now exercised.
    let branch_summary_msg = session
        .messages
        .iter()
        .find(|m| m.metadata.get("pi_type").map(String::as_str) == Some("branch_summary"))
        .expect("the active-path branch_summary must map to a canonical message");
    assert_eq!(branch_summary_msg.role, Role::User);
    assert_eq!(
        branch_summary_msg.content.as_deref(),
        Some(
            "[branch summary]\ndrafted a longer confirmation reply, then rewound and kept \
             the terse one"
        ),
        "the branch_summary's `summary` text must become the canonical User message body"
    );
    assert_eq!(
        branch_summary_msg
            .metadata
            .get("pi_from_id")
            .map(String::as_str),
        Some("e4b00004"),
        "`fromId` must carry through as `pi_from_id` metadata"
    );
    assert!(
        !branch_summary_msg.metadata.contains_key("compacted_out"),
        "the branch_summary itself must never be marked compacted_out"
    );

    // SessionMeta: model = last assistant/model_change on the active path
    // (here, the final assistant turn's own model wins over the earlier
    // model_change record).
    assert_eq!(session.meta.model.as_deref(), Some("claude-opus-4-6"));
    assert_eq!(
        session.meta.session_id.as_deref(),
        Some("1e6f2a3b-0000-4000-8000-000000000001")
    );
    assert_eq!(session.meta.cwd, Some(PathBuf::from("/tmp/demo-repo")));
    assert_eq!(session.meta.system_prompt, None);
    assert_eq!(
        session.meta.lineage.get("session_name").map(String::as_str),
        Some("demo session")
    );
    assert_eq!(
        session.meta.lineage.get("pi_version").map(String::as_str),
        Some("3")
    );
}

#[test]
fn pi_tool_result_name_survives_claude_and_codex_round_trips() {
    let source = Session::from_pi(fixture("pi_real_corpus_tool_call.jsonl")).unwrap();
    assert_pi_tool_result_names_survive_foreign_round_trips(&source);
}

fn assert_pi_tool_result_names_survive_foreign_round_trips(source: &Session) {
    let source_tools = source
        .messages
        .iter()
        .filter(|message| message.role == Role::Tool)
        .map(|message| (message.tool_call_id.clone(), message.name.clone()))
        .collect::<Vec<_>>();
    assert_eq!(source_tools.len(), 2, "fixture has two tool results");

    for intermediate in [SessionFormat::ClaudeCode, SessionFormat::Codex] {
        let exported = source.to_jsonl(intermediate).unwrap();
        let loaded = Session::load_str(&exported, intermediate).unwrap();
        let returned = loaded.to_jsonl(SessionFormat::Pi).unwrap();
        let restored = Session::from_pi_str(&returned).unwrap();
        let restored_tools = restored
            .messages
            .iter()
            .filter(|message| message.role == Role::Tool)
            .map(|message| (message.tool_call_id.clone(), message.name.clone()))
            .collect::<Vec<_>>();

        assert_eq!(restored_tools, source_tools, "via {intermediate:?}");
    }
}

// ---- (c) completeness guard (§4.1, S6) -------------------------------------

#[test]
fn pi_fixture_has_no_unknown_records_or_roles() {
    let tmp = std::env::temp_dir().join(format!("sc-pi-cov-{}-{}", std::process::id(), "fixture"));
    std::fs::create_dir_all(&tmp).unwrap();
    std::fs::copy(fixture("pi_session.jsonl"), tmp.join("pi_session.jsonl")).unwrap();

    let report = audit_dir(&tmp, Corpus::Pi, None);
    assert_eq!(report.parse_errors, 0, "pi fixture: parse errors");

    let unknown: Vec<_> = report
        .records
        .keys()
        .filter(|k| k.starts_with("<line>/"))
        .collect();
    assert!(
        unknown.is_empty(),
        "pi fixture: unknown record discriminants: {unknown:?}"
    );

    // S6: the second-level `message.role` discriminant's UnknownRole bucket
    // must be empty over the committed fixture.
    let unknown_roles: Vec<_> = report
        .records
        .keys()
        .filter(|k| k.starts_with("message/UnknownRole"))
        .collect();
    assert!(
        unknown_roles.is_empty(),
        "pi fixture: UnknownRole bucket must be empty: {unknown_roles:?}"
    );

    // FIX #2: the second-level `UnknownImageShape` bucket must likewise be
    // empty — the fixture's one image (`eimg0001`) matches the assumed
    // `{mimeType, data}` shape, so it must map normally, not fail the guard.
    let unknown_image_shapes: Vec<_> = report
        .records
        .keys()
        .filter(|k| k.starts_with("message/UnknownImageShape"))
        .collect();
    assert!(
        unknown_image_shapes.is_empty(),
        "pi fixture: UnknownImageShape bucket must be empty: {unknown_image_shapes:?}"
    );

    std::fs::remove_dir_all(&tmp).ok();
}

/// S6: pi's `message.role` is an OPEN, extension-mergeable union. An unknown
/// role must (1) survive in `raw` (never dropped/panicked on), (2) produce
/// NO canonical message, and (3) make the coverage guard's `UnknownRole`
/// bucket non-empty — the guard's entire purpose is to turn this into a
/// visible failure instead of a silent drop.
#[test]
fn pi_unknown_role_is_raw_only_and_trips_the_coverage_guard() {
    let jsonl = concat!(
        r#"{"type":"session","version":3,"id":"s1","timestamp":"2026-01-01T00:00:00.000Z","cwd":"/tmp"}"#,
        "\n",
        r#"{"type":"message","id":"a1","parentId":null,"timestamp":"2026-01-01T00:00:01.000Z","message":{"role":"user","content":"hi","timestamp":1}}"#,
        "\n",
        r#"{"type":"message","id":"a2","parentId":"a1","timestamp":"2026-01-01T00:00:02.000Z","message":{"role":"futureRole","content":"from an extension pi doesn't ship","timestamp":2}}"#,
    );
    let session = Session::from_pi_str(jsonl).unwrap();

    assert!(
        session.raw.iter().any(|l| l.contains("futureRole")),
        "the unknown-role line must survive verbatim in raw"
    );
    assert_eq!(
        session.messages.len(),
        1,
        "an unknown role must produce NO canonical message (raw-only survival)"
    );

    let tmp = std::env::temp_dir().join(format!("sc-pi-unknown-{}", std::process::id()));
    std::fs::create_dir_all(&tmp).unwrap();
    std::fs::write(tmp.join("s.jsonl"), jsonl).unwrap();
    let report = audit_dir(&tmp, Corpus::Pi, None);
    let unknown_roles: Vec<_> = report
        .records
        .keys()
        .filter(|k| k.starts_with("message/UnknownRole"))
        .collect();
    assert!(
        !unknown_roles.is_empty(),
        "the coverage guard must flag the unknown role, not silently pass"
    );
    std::fs::remove_dir_all(&tmp).ok();
}

/// FIX #2: an `ImageContent` block whose shape doesn't match the loader's
/// (unverified) `{mimeType, data}` guess must NEVER be silently synthesized
/// as an empty/corrupt `image_url` part. It must (1) survive verbatim in
/// `raw`, (2) produce NO canonical message for the message it's in (raw-only
/// survival, exactly like an unknown `message.role`), and (3) make the
/// coverage guard's `UnknownImageShape` bucket non-empty.
#[test]
fn pi_unknown_image_shape_is_raw_only_and_trips_the_coverage_guard() {
    let jsonl = concat!(
        r#"{"type":"session","version":3,"id":"s1","timestamp":"2026-01-01T00:00:00.000Z","cwd":"/tmp"}"#,
        "\n",
        r#"{"type":"message","id":"a1","parentId":null,"timestamp":"2026-01-01T00:00:01.000Z","message":{"role":"user","content":"hi","timestamp":1}}"#,
        "\n",
        // A "message/user" whose content carries an image block with an
        // unexpected shape: no `data` field at all (e.g. a future pi version
        // that ships images as `{type:"image", source:{...}}` instead) —
        // exactly the "real pi image with a different shape" failure mode
        // FIX #2 targets. Must NOT be defaulted to mimeType="image/png",
        // data="".
        r#"{"type":"message","id":"a2","parentId":"a1","timestamp":"2026-01-01T00:00:02.000Z","message":{"role":"user","content":[{"type":"text","text":"see attached"},{"type":"image","source":{"kind":"file","path":"/tmp/screenshot.png"}}],"timestamp":2}}"#,
    );
    let session = Session::from_pi_str(jsonl).unwrap();

    assert!(
        session.raw.iter().any(|l| l.contains("screenshot.png")),
        "the malformed-image-shape line must survive verbatim in raw"
    );
    assert_eq!(
        session.messages.len(),
        1,
        "a message containing an unrecognized image shape must produce NO canonical \
         message (raw-only survival) — never a synthesized empty/corrupt image part"
    );
    assert!(
        !session.messages.iter().any(|m| m.content_parts.is_some()),
        "no corrupt image_url part may be synthesized from the unrecognized shape"
    );

    let tmp = std::env::temp_dir().join(format!("sc-pi-unknown-image-{}", std::process::id()));
    std::fs::create_dir_all(&tmp).unwrap();
    std::fs::write(tmp.join("s.jsonl"), jsonl).unwrap();
    let report = audit_dir(&tmp, Corpus::Pi, None);
    let unknown_image_shapes: Vec<_> = report
        .records
        .keys()
        .filter(|k| k.starts_with("message/UnknownImageShape"))
        .collect();
    assert!(
        !unknown_image_shapes.is_empty(),
        "the coverage guard must flag the unrecognized image shape, not silently pass"
    );
    std::fs::remove_dir_all(&tmp).ok();
}

// ---- (d) writer: version:3 + header <=512 bytes ----------------------------

#[test]
fn pi_writer_emits_v3_header_within_512_bytes() {
    let session = Session::from_pi(fixture("pi_session.jsonl")).unwrap();
    let out = session.to_jsonl(SessionFormat::Pi).unwrap();

    let header_line = out.lines().next().expect("writer must emit a header line");
    let header: serde_json::Value = serde_json::from_str(header_line).unwrap();
    assert_eq!(header.get("type").and_then(|v| v.as_str()), Some("session"));
    assert_eq!(
        header.get("version").and_then(|v| v.as_i64()),
        Some(3),
        "pi writer must always normalize/emit version 3"
    );
    assert!(
        header_line.len() <= 512,
        "pi header line must stay <=512 bytes (readSessionHeader only reads the \
         first 512, S10): got {} bytes",
        header_line.len()
    );

    for line in out.lines().skip(1).filter(|l| !l.trim().is_empty()) {
        let v: serde_json::Value = serde_json::from_str(line)
            .unwrap_or_else(|e| panic!("writer produced invalid JSON line ({e}): {line}"));
        assert!(v.get("type").is_some(), "entry missing `type`: {line}");
        assert!(v.get("id").is_some(), "entry missing `id`: {line}");
        assert!(
            v.get("timestamp").is_some(),
            "entry missing `timestamp`: {line}"
        );
        assert!(
            v.as_object().is_some_and(|o| o.contains_key("parentId")),
            "entry missing `parentId`: {line}"
        );
    }

    // Reload sanity — a view round-trip (T3), not claimed byte-lossless: the
    // synthesized file still loads back into a sensible conversation.
    let reloaded = Session::from_pi_str(&out).unwrap();
    assert!(!reloaded.messages.is_empty());
    assert_eq!(reloaded.meta.session_id, session.meta.session_id);
}

// ---- (e) splice: prefix-verbatim --------------------------------------------

#[test]
fn pi_splice_replays_the_imported_prefix_verbatim() {
    let session = Session::from_pi(fixture("pi_session.jsonl")).unwrap();
    let raw_prefix_before = session.raw.clone();

    // Route the appended turns through the native-v2 sidecar mechanism
    // (mirrors `spliced_export.rs`'s pattern exactly): that's what keeps
    // `raw`/`messages` growing in lockstep for the appended tail, which is
    // what `Session::spliced_prefix_lens` relies on to find the splice
    // boundary. Directly pushing onto `session.messages` in place, with no
    // corresponding `raw` line, is not how a live agent loop appends turns —
    // it goes through the sidecar (D1).
    let appended = vec![
        ChatMessage::user("one more thing"),
        ChatMessage::assistant("sure thing"),
    ];
    let sidecar = session.to_native_jsonl_v2(&appended);
    let reconstructed = Session::from_sidecar_str(&sidecar).unwrap();

    let spliced = reconstructed
        .to_jsonl_spliced(SessionFormat::Pi, None)
        .unwrap();
    let spliced_lines: Vec<&str> = spliced.lines().collect();

    assert!(spliced_lines.len() > raw_prefix_before.len());
    for (i, line) in raw_prefix_before.iter().enumerate() {
        assert_eq!(
            &spliced_lines[i], line,
            "splice must replay raw line {i} verbatim (already-v3 header, no id override)"
        );
    }

    let appended = spliced_lines[raw_prefix_before.len()..].join("\n");
    assert!(appended.contains("one more thing"));
    assert!(appended.contains("sure thing"));

    let reloaded = Session::from_pi_str(&spliced).unwrap();
    assert!(reloaded
        .messages
        .iter()
        .any(|m| m.content.as_deref() == Some("one more thing")));
    assert!(reloaded
        .messages
        .iter()
        .any(|m| m.content.as_deref() == Some("sure thing")));

    println!(
        "pi splice: {} raw prefix lines replayed verbatim, {} turns appended — PASS",
        raw_prefix_before.len(),
        2
    );
}

/// A splice with a `session_id` override must rewrite the header's `id`
/// (and the id must not go through a re-parse for any other raw line — pi
/// repeats the session id on no other line).
#[test]
fn pi_splice_rewrites_session_id_only_on_the_header() {
    let session = Session::from_pi(fixture("pi_session.jsonl")).unwrap();
    let spliced = session
        .to_jsonl_spliced(SessionFormat::Pi, Some("new-session-id"))
        .unwrap();
    let mut lines = spliced.lines();
    let header: serde_json::Value = serde_json::from_str(lines.next().unwrap()).unwrap();
    assert_eq!(
        header.get("id").and_then(|v| v.as_str()),
        Some("new-session-id")
    );
    assert_eq!(header.get("version").and_then(|v| v.as_i64()), Some(3));

    // Every other raw line is untouched — none of them carry the new id.
    for line in spliced.lines().skip(1).take(session.raw.len() - 1) {
        assert!(!line.contains("new-session-id"));
    }
}

// ---- msg_eq_multimodal sanity (S2) -----------------------------------------

/// The multimodal comparator must actually distinguish a message that kept
/// its image from one that (hypothetically) dropped it — otherwise it's no
/// stronger than the base `msg_eq` it's meant to fix.
#[test]
fn msg_eq_multimodal_is_not_blind_to_dropped_images() {
    let with_image = ChatMessage {
        role: Role::User,
        content: None,
        content_parts: Some(vec![
            serde_json::json!({"type": "text", "text": "look"}),
            serde_json::json!({"type": "image_url", "image_url": {"url": "data:image/png;base64,aGVsbG8="}}),
        ]),
        tool_calls: None,
        tool_call_id: None,
        name: None,
        metadata: Default::default(),
    };
    let without_image = ChatMessage {
        content_parts: Some(vec![serde_json::json!({"type": "text", "text": "look"})]),
        ..with_image.clone()
    };
    assert!(
        !msg_eq_multimodal(&with_image, &without_image),
        "msg_eq_multimodal must NOT equate a message with an image to one without it"
    );
    assert!(msg_eq_multimodal(&with_image, &with_image.clone()));
}

// ---- PARITY-2 / IX-2: real earendil-works/pi corpus validation -------------

/// `pi_session_live_corpus.jsonl` is not hand-authored: it was produced by
/// importing pi's OWN `SessionManager` class straight from a shallow clone of
/// `earendil-works/pi` @351efc828b6fc5250fa50d6b32b20b0f0cb22cb4 (the pin
/// `pi-fields.md` cites) and driving it through `SessionManager.create()` +
/// `appendMessage()`/`appendSessionInfo()` — the exact code path the real
/// `pi` CLI uses to persist a session, just without a live LLM call (this box
/// has no model API key). The file landed at pi's real default location,
/// `~/.pi/agent/sessions/--<encoded-cwd>--/<timestamp>_<uuid>.jsonl`, and was
/// re-opened successfully through pi's own `SessionManager.open()` (no
/// version-migration rewrite — it was already emitted as v3) before being
/// copied here verbatim. This closes the corpus-validation gap
/// `docs/interop/build-followups.md` IX-2 flagged as open ("what's missing is
/// corpus access to a real earendil-works/pi session").
#[test]
fn pi_live_corpus_round_trips_and_matches_the_spec_exactly() {
    let path = fixture("pi_session_live_corpus.jsonl");
    let original_bytes = std::fs::read(&path).unwrap();

    let session = Session::from_pi(&path).unwrap();

    // Header fields (SessionHeader, sm:32-39 in pi-fields.md's citation
    // scheme) came through exactly as pi's own `newSession()` wrote them.
    assert_eq!(
        session.meta.session_id.as_deref(),
        Some("019f401c-658f-7a34-a6df-48e7eade79c4")
    );
    assert_eq!(
        session.meta.cwd,
        Some(PathBuf::from("/tmp/demo-pi-project"))
    );
    assert_eq!(
        session.meta.lineage.get("pi_version").map(String::as_str),
        Some("3")
    );

    // Every record type pi's real SessionManager emitted in this run
    // round-trips into canonical messages: user, assistant+toolCall,
    // toolResult, bashExecution, a second assistant close, and the trailing
    // session_info (no canonical message, correctly — see pi-fields.md §11).
    assert_eq!(
        session.messages.len(),
        6,
        "user, 3x assistant, toolResult, bashExecution: {:?}",
        session.messages
    );

    let call = session
        .messages
        .iter()
        .find(|m| !m.tool_calls().is_empty())
        .expect("real pi ToolCall entry")
        .tool_calls()[0]
        .clone();
    assert_eq!(call.function.name, "write_file");
    assert_eq!(
        call.function.parsed_arguments().unwrap(),
        serde_json::json!({"path": "hello.txt", "content": "hi"})
    );

    let tool_result = session
        .messages
        .iter()
        .find(|m| m.role == Role::Tool)
        .expect("real pi toolResult entry");
    assert_eq!(tool_result.tool_call_id.as_deref(), Some(call.id.as_str()));
    assert_eq!(
        tool_result.content.as_deref(),
        Some("wrote 2 bytes to hello.txt")
    );

    let bash = session
        .messages
        .iter()
        .find(|m| m.metadata.contains_key("pi_bash_command"))
        .expect("real pi bashExecution entry");
    assert_eq!(
        bash.metadata.get("pi_bash_command").map(String::as_str),
        Some("cat hello.txt")
    );
    assert_eq!(
        bash.metadata.get("pi_bash_output").map(String::as_str),
        Some("hi")
    );

    // Coverage guard: zero unknown records/roles over genuine pi output —
    // the strongest form of the S6 completeness claim, since this input was
    // never shaped by hand to satisfy the loader.
    let tmp = std::env::temp_dir().join(format!("sc-pi-live-cov-{}", std::process::id()));
    std::fs::create_dir_all(&tmp).unwrap();
    std::fs::copy(&path, tmp.join("pi_session_live_corpus.jsonl")).unwrap();
    let report = audit_dir(&tmp, Corpus::Pi, None);
    assert_eq!(report.parse_errors, 0);
    assert!(
        report.records.keys().all(|k| !k.starts_with("<line>/")
            && !k.contains("UnknownRole")
            && !k.contains("UnknownImageShape")),
        "genuine pi output must hit zero unknown-record/unknown-role/unknown-image-shape \
         buckets: {:?}",
        report.records.keys().collect::<Vec<_>>()
    );
    std::fs::remove_dir_all(&tmp).ok();

    // Byte-lossless native round-trip, same T1-byte proof as the hand-authored
    // fixture above, now against genuine pi-emitted bytes.
    let native = session.to_native_jsonl();
    let reconstructed_body = strip_native_header(&native);
    assert_eq!(
        reconstructed_body.as_bytes(),
        original_bytes.as_slice(),
        "native round-trip must reproduce the genuine pi corpus file byte-for-byte"
    );

    println!(
        "pi LIVE CORPUS ({} bytes, produced by real earendil-works/pi@351efc8 \
         SessionManager code, reopened clean via SessionManager.open()): \
         6/6 messages canonicalized, tool call+result paired, bashExecution fields \
         preserved, 0 unknown-record/role/image-shape, byte-lossless round-trip — PASS",
        original_bytes.len()
    );
}

// ---- (f) B4: content-bearing Claude `system` message survives the pi hop --

/// B4 (priority-1 losslessness gap, surfaced against `push_claude_system`'s
/// keep-listed subtypes — `local_command`/`scheduled_task_fire`/
/// `away_summary`): a content-bearing `Role::System` `ChatMessage` (real
/// text, e.g. a `<local-command-stdout>...</local-command-stdout>` record,
/// NOT the session-level system prompt, which correctly has no pi slot and
/// stays excluded) must not be silently dropped writing to pi.
/// `write_pi_entries`'s `Role::System` arm used to unconditionally
/// `continue`; it now re-materializes the record as a pi `role:"custom"`
/// message (§3e, the closest existing non-fabricated pi slot for
/// "extension-injected, sent to the LLM as a user message"), tagged with the
/// `supercode_claude_system` marker `customType` + a `details.
/// claude_system_subtype` channel, and `push_pi_custom_common` restores it
/// back to `Role::System` + `metadata["systemSubtype"]` on reload — a real
/// Claude -> Pi -> (reload) round trip, not just a one-way emit.
///
/// Non-vacuous: FAILS against `parity/integrated-v3@e7b15fd` (the
/// `Role::System => continue` arm drops the message outright — reloading the
/// pi output yields ZERO `Role::System` messages, not one) and PASSES here.
#[test]
fn pi_writer_preserves_content_bearing_claude_system_message() {
    let mut session = Session::from_claude_code_str("").unwrap();
    let sys_content =
        "<local-command-stdout>Not enough messages to compact.</local-command-stdout>";
    session.messages = vec![
        ChatMessage::user("run /compact"),
        ChatMessage::system(sys_content)
            .with_meta("systemSubtype", "local_command")
            .with_meta("timestamp", "2026-07-06T13:10:14.575Z"),
        ChatMessage::assistant("noted"),
    ];

    let out = session.to_jsonl(SessionFormat::Pi).unwrap();
    assert!(
        out.contains("supercode_claude_system"),
        "pi writer must emit the recognizable custom-message marker for a \
         content-bearing System record, not drop it silently"
    );

    let reloaded = Session::from_pi_str(&out).unwrap();
    let sys_msgs: Vec<_> = reloaded
        .messages
        .iter()
        .filter(|m| m.role == Role::System)
        .collect();
    assert_eq!(
        sys_msgs.len(),
        1,
        "exactly one Role::System message must survive Claude -> Pi -> reload \
         (got {} — the record was dropped or duplicated): messages = {:?}",
        sys_msgs.len(),
        reloaded
            .messages
            .iter()
            .map(|m| &m.role)
            .collect::<Vec<_>>()
    );
    assert_eq!(
        sys_msgs[0].content.as_deref(),
        Some(sys_content),
        "the ORIGINAL text must be conserved verbatim, never fabricated/altered"
    );
    assert_eq!(
        sys_msgs[0]
            .metadata
            .get("systemSubtype")
            .map(String::as_str),
        Some("local_command"),
        "the original Claude subtype must be recoverable from the pi leg, \
         not lost or guessed wrong"
    );

    // Sanity: the other two messages (user/assistant) must also survive
    // unaffected — this fix must not perturb the surrounding conversation.
    assert_eq!(reloaded.messages.len(), 3);
    assert_eq!(reloaded.messages[0].role, Role::User);
    assert_eq!(reloaded.messages[2].role, Role::Assistant);
}