clustervision 0.16.1

//! `cv show` (and its `--tree` / `--kind` views) / `cv export` — reading sessions,
//! plus the streaming renderer (`stream_session_render`) and its header/message helpers.

use crate::util::{
    clamp, continue_hint, count_messages, home_rel, parse_harness, resolve, resolve_found, short_id, split_harness_id,
    usage, Window, WindowArgs,
};
use anyhow::{bail, Context, Result};
use cv_core::ir::{truncate, Block, Harness, Message, MessageKind, Role, Session, SessionRef};
use cv_core::render::turn_label;
use cv_core::Adapter;
use std::io::{IsTerminal, Write};
use std::path::PathBuf;

/// A piped `cv show` with no selector prints at most this much before switching to head + tail:
/// agents read cv through pipes constantly, and a full transcript in a tool result is the
/// context-window failure mode this exists to prevent.
const PIPE_GUARD_BYTES: usize = 200 * 1024;
/// How many messages each end of the head + tail view shows.
const PIPE_GUARD_EDGE: usize = 20;

/// The views `cv show` offers instead of the plain transcript (each flag documented on `Cmd::Show`).
pub(crate) struct ShowView {
    pub subagents: bool,
    pub agent: Option<String>,
    pub pre_compaction: Option<usize>,
    pub tree: bool,
    pub kind: Option<String>,
    pub summaries: bool,
}

/// What `--kind` selects: each family has one compact view.
enum KindView {
    /// `prompt` / `answer`: what the person said (`orchestrate::print_prompts`).
    Said { prompts: bool, answers: bool },
    /// `compaction_boundary` / `compaction_summary`: the compaction seams, each boundary paired
    /// with its summary (`compaction::print_compactions`).
    Compactions,
    /// `plan`, `schedule`, `background_job`, … (or `affordances`, all of them): the affordance rows
    /// (`cv_core::affordance`, `docs/design/AFFORDANCES.md`).
    Affordances(std::collections::HashSet<cv_core::affordance::Kind>),
}

fn parse_kinds(spec: &str) -> Result<KindView> {
    const SAID: [&str; 2] = ["prompt", "answer"];
    const SEAMS: [&str; 2] = ["compaction_boundary", "compaction_summary"];
    let kinds: Vec<&str> = spec.split(',').map(str::trim).filter(|k| !k.is_empty()).collect();
    let words: Vec<&str> = cv_core::affordance::Kind::ALL.iter().map(|k| k.as_str()).collect();
    let accepted = format!(
        "`--kind` takes `prompt,answer` (what the person said), \
         `compaction_boundary,compaction_summary` (the compaction seams), or affordances: {} (`affordances` is all of them)",
        words.join(",")
    );
    let family = |k: &str| {
        if SAID.contains(&k) {
            Some(0)
        } else if SEAMS.contains(&k) {
            Some(1)
        } else if k == "affordances" || cv_core::affordance::Kind::parse(k).is_some() {
            Some(2)
        } else {
            None
        }
    };
    if let Some(k) = kinds.iter().find(|k| family(k).is_none()) {
        return usage(format!("no `--kind` view for {k:?} — {accepted}"));
    }
    let families: std::collections::BTreeSet<u8> = kinds.iter().filter_map(|k| family(k)).collect();
    match families.iter().copied().collect::<Vec<_>>().as_slice() {
        [] => usage(format!("`--kind` is empty — {accepted}")),
        [0] => Ok(KindView::Said {
            prompts: kinds.contains(&"prompt"),
            answers: kinds.contains(&"answer"),
        }),
        [1] => Ok(KindView::Compactions),
        [2] => Ok(KindView::Affordances(
            cv_core::affordance::parse_kinds(&kinds).expect("every word checked above"),
        )),
        _ => usage(format!("`--kind {spec}` mixes two views — {accepted}")),
    }
}

/// `show --kind plan,schedule,…`: the affordance rows of one session, transcript rows in message
/// order then the side store's. A window narrows the transcript rows; store rows always print.
fn print_affordances(
    r: &SessionRef,
    adapter: &dyn Adapter,
    kinds: &std::collections::HashSet<cv_core::affordance::Kind>,
    range: Option<(usize, Option<usize>)>,
    json: bool,
) -> Result<()> {
    use cv_core::sanitize::sanitize_line;
    let mut rows = cv_core::affordance::read(adapter, r)?;
    rows.retain(|a| {
        kinds.contains(&a.kind)
            && match (a.index, range) {
                (Some(i), Some((start, end))) => i >= start && end.is_none_or(|e| i < e),
                _ => true,
            }
    });
    if json {
        println!("{}", serde_json::to_string_pretty(&rows)?);
        return Ok(());
    }
    println!("# affordances of {} — {} row(s)\n", short_id(&r.id), rows.len());
    for a in &rows {
        let at = match (a.index, a.source) {
            (Some(i), _) => format!("[{i}]"),
            (None, cv_core::affordance::Source::Store) => "store".to_string(),
            (None, _) => "session".to_string(),
        };
        let ts = a
            .timestamp
            .map(|t| crate::util::fmt_local(t, "%Y-%m-%d %H:%M"))
            .unwrap_or_else(|| "----------------".into());
        let tool = a.tool.as_deref().map(|t| format!("  {t}")).unwrap_or_default();
        println!(
            "{at} {ts}  {}/{}{}  {}",
            a.kind.as_str(),
            a.action,
            sanitize_line(&tool),
            sanitize_line(&a.summary)
        );
        for it in &a.items {
            let st = it.status.as_deref().unwrap_or("-");
            println!(
                "      [{}] {}",
                sanitize_line(st),
                sanitize_line(&truncate(&it.content, 160))
            );
        }
    }
    Ok(())
}

pub(crate) fn cmd_show(
    id: &str,
    harness: Option<String>,
    json: bool,
    window: &WindowArgs,
    view: ShowView,
) -> Result<()> {
    let ShowView {
        subagents,
        agent,
        pre_compaction,
        tree,
        kind,
        summaries,
    } = view;
    // Parse the selector (and `--kind`) before any I/O so a bad value errors immediately.
    let selector = window.window()?;
    let kind = kind.as_deref().map(parse_kinds).transpose()?;
    if summaries && !matches!(kind, Some(KindView::Compactions)) {
        return usage("`--summaries` goes with `--kind compaction_boundary,compaction_summary`");
    }

    // The compact views read a cataloged session (never a sub-agent id) over the whole session,
    // except that what the person said can be narrowed to a window.
    if let Some(kind) = kind {
        let (r, adapter) = resolve(id, parse_harness(&harness)?)?;
        return match kind {
            KindView::Said { prompts, answers } => {
                if window.max_bytes.is_some() {
                    return usage("`--max-bytes` cuts a transcript; `--kind prompt,answer` rows are printed whole");
                }
                let range = show_range(&r, window, pre_compaction)?
                    .map(|w| w.bounds(|| count_messages(adapter.as_ref(), &r)))
                    .transpose()?;
                crate::cmd::orchestrate::print_prompts(&r, adapter.as_ref(), range, prompts, answers, json)
            }
            KindView::Affordances(kinds) => {
                if window.max_bytes.is_some() {
                    return usage("`--max-bytes` cuts a transcript; the affordance rows are printed whole");
                }
                let range = show_range(&r, window, pre_compaction)?
                    .map(|w| w.bounds(|| count_messages(adapter.as_ref(), &r)))
                    .transpose()?;
                print_affordances(&r, adapter.as_ref(), &kinds, range, json)
            }
            KindView::Compactions => {
                if selector.is_some() || window.max_bytes.is_some() || pre_compaction.is_some() {
                    return usage("the compaction view covers the whole session; drop the window flags");
                }
                crate::cmd::compaction::print_compactions(&r, summaries, json)
            }
        };
    }
    if tree && !json {
        if selector.is_some() || window.max_bytes.is_some() {
            return usage("`--tree` renders the whole session; drop the window flags (or add `--json` for the IR)");
        }
        let (r, adapter) = resolve(id, parse_harness(&harness)?)?;
        return print_tree(&r, adapter.as_ref());
    }

    let (want, id) = split_harness_id(id, parse_harness(&harness)?);

    // An `agent-…` shaped id goes straight to the sub-agent view: cv_core::find_cheap can now
    // resolve these too (its fleet-wide fallback), but this path keeps the provenance banner and
    // lists candidates on an ambiguous prefix instead of erroring.
    if id.starts_with("agent-") && show_agent_fleetwide(id, json, window)? {
        return Ok(());
    }
    // find_cheap first: don't pay a full fleet re-discovery before trying the id as a sub-agent
    // id — sub-agents aren't in the main pool, but `cv show <agent-id>` should still just work
    // (harvest reports hand out bare agent ids all the time). Only when the id is neither a
    // cataloged session nor an agent does the full-scan `find` escalation run (a session in the
    // discovery probe's blind spots).
    let found = match cv_core::find_cheap(id, want) {
        Ok(Some(hit)) => Ok(Some(hit)),
        Ok(None) => {
            if show_agent_fleetwide(id, json, window)? {
                return Ok(());
            }
            cv_core::find(id, want)
        }
        Err(e) => Err(e),
    };
    let (r, adapter) = match found {
        Ok(Some(hit)) => hit,
        Ok(None) => return usage(format!("no session (and no sub-agent) matching {id:?}")),
        Err(e) => return resolve_found(Err(e), id, want).map(|_| ()),
    };

    // `--agent <id>`: render one specific sub-agent's transcript (resolved through this parent,
    // since sub-agents aren't in the main pool). `--subagents`: list the whole forest with results.
    if let Some(agent_id) = &agent {
        return show_one_subagent(&r, adapter.as_ref(), agent_id, json, window);
    }
    if subagents {
        return show_subagents(&r, json);
    }

    let range = show_range(&r, window, pre_compaction)?;
    if json {
        return print_session_json(adapter.as_ref(), &r, range, window.max_bytes, false, None).map(|_| ());
    }
    // The pipe guard applies only to a selector-less, budget-less render (an explicit window or
    // budget is the caller saying what they want).
    let guard = selector.is_none() && pre_compaction.is_none() && window.max_bytes.is_none();
    render_text(adapter.as_ref(), &r, range, window.max_bytes, guard)
}

/// The window `show` reads: the selector's, or — under `--pre-compaction <N>` — the span before
/// the Nth (1-based) compaction boundary, the context the continued agent lost, so it is
/// retrieved by message range without the caller computing offsets by hand. Left unresolved:
/// `--last N` is resolved by whichever pass reads the session, not by a counting pass first.
fn show_range(r: &SessionRef, window: &WindowArgs, pre_compaction: Option<usize>) -> Result<Option<Window>> {
    let Some(n) = pre_compaction else {
        return window.window();
    };
    let comps = cv_core::compaction::detect(r, false)?;
    if comps.is_empty() {
        bail!("{} never compacted — nothing pre-compaction to show", short_id(&r.id));
    }
    let idx = n.saturating_sub(1);
    let (start, end) = cv_core::compaction::pre_compaction_span(&comps, idx).with_context(|| {
        format!(
            "{} compacted {} time(s); no compaction #{n} (use 1..={})",
            short_id(&r.id),
            comps.len(),
            comps.len(),
        )
    })?;
    eprintln!(
        "✦ pre-compaction #{n} of {}: messages {start}..{end} (the span before boundary @msg {})",
        comps.len(),
        comps[idx].boundary_msg_idx,
    );
    Ok(Some(Window::Range(start, Some(end))))
}

/// `show --json` / `export --format json`: the whole IR (incl. `extra`), windowed, and — under
/// `--max-bytes` — cut at the budget with the continuation hint on **stderr** (stdout stays JSON).
///
/// `as_document` wraps the same bytes as an **OpenSession** document (`docs/OPENSESSION.md`):
/// the IR's own serialization with `"open_session": "0.3"` as the first key. That marker belongs
/// to a *document*, not to a `Session` — `cv show --json`, the MCP payloads and `cv serve` all serialize
/// the raw IR and their key sets are pinned — so it is added here, at the one boundary where cv
/// hands a session to somebody else, and nowhere else.
fn print_session_json(
    adapter: &dyn Adapter,
    r: &SessionRef,
    window: Option<Window>,
    max_bytes: Option<usize>,
    as_document: bool,
    redact: Option<&cv_core::redact::RedactOptions>,
) -> Result<cv_core::redact::RedactStats> {
    let mut session = adapter.parse(r)?;
    let range = window.map(|w| w.bounds(|| Ok(session.messages.len()))).transpose()?;
    let mut stats = cv_core::redact::RedactStats::default();
    let start = range.map(|(s, _)| s).unwrap_or(0);
    if let Some(rg) = range {
        let (s, e) = clamp(rg, session.messages.len());
        session.messages = session.messages.drain(s..e).collect();
    }
    if let Some(budget) = max_bytes {
        let mut bytes = 0usize;
        let mut keep = 0usize;
        for (i, m) in session.messages.iter().enumerate() {
            let n = serde_json::to_vec(m)?.len();
            if keep > 0 && bytes + n > budget {
                break;
            }
            bytes += n;
            keep = i + 1;
        }
        if keep < session.messages.len() {
            eprintln!("{}", continue_hint(start + keep));
            session.messages.truncate(keep);
        }
    }
    if let Some(opts) = redact {
        let (clean, st) = cv_core::redact::redact_with(&session, opts);
        session = clean;
        stats = st;
    }
    if as_document {
        println!(
            "{}",
            serde_json::to_string_pretty(&cv_core::harness::opensession::document(&session))?
        );
    } else {
        println!("{}", serde_json::to_string_pretty(&session)?);
    }
    Ok(stats)
}

/// The rendered-transcript path of `cv show`, with the pipe guard: a selector-less render to a
/// non-terminal stdout that would exceed [`PIPE_GUARD_BYTES`] becomes the first and last
/// [`PIPE_GUARD_EDGE`] messages with a hint line between them.
fn render_text(
    adapter: &dyn Adapter,
    r: &SessionRef,
    range: Option<Window>,
    max_bytes: Option<usize>,
    guard: bool,
) -> Result<()> {
    let stdout = std::io::stdout();
    if guard && !stdout.is_terminal() {
        // Render into memory under the guard's budget; if it all fits, that IS the output.
        let mut buf = Vec::new();
        let probe = stream_session_render(
            adapter,
            r,
            &mut buf,
            show_header,
            show_message,
            RenderOpts {
                range: None,
                max_bytes: Some(PIPE_GUARD_BYTES),
            },
        )?;
        let mut out = std::io::BufWriter::new(stdout.lock());
        if probe.next.is_none() {
            out.write_all(&buf)?;
            out.flush()?;
            return Ok(());
        }
        let total = count_messages(adapter, r)?;
        if total <= 2 * PIPE_GUARD_EDGE {
            // Too few messages to elide any — the size is in the messages themselves.
            stream_session_render(adapter, r, &mut out, show_header, show_message, RenderOpts::default())?;
            out.flush()?;
            return Ok(());
        }
        stream_session_render(
            adapter,
            r,
            &mut out,
            show_header,
            show_message,
            RenderOpts {
                range: Some(Window::Range(0, Some(PIPE_GUARD_EDGE))),
                max_bytes: None,
            },
        )?;
        writeln!(
            out,
            "… {} messages omitted ({total} total; the full render exceeds {} KB on a pipe) — read them with \
             --first N, --last N, --range A..B, or --around N; --max-bytes N caps output\n",
            total - 2 * PIPE_GUARD_EDGE,
            PIPE_GUARD_BYTES / 1024,
        )?;
        stream_session_render(
            adapter,
            r,
            &mut out,
            |_| String::new(),
            show_message,
            RenderOpts {
                range: Some(Window::Range(total - PIPE_GUARD_EDGE, None)),
                max_bytes: None,
            },
        )?;
        out.flush()?;
        return Ok(());
    }

    let mut out = std::io::BufWriter::new(stdout.lock());
    let outcome = stream_session_render(
        adapter,
        r,
        &mut out,
        show_header,
        show_message,
        RenderOpts { range, max_bytes },
    )?;
    if let Some(next) = outcome.next {
        writeln!(out, "{}", continue_hint(next))?;
    }
    out.flush()?;
    Ok(())
}

/// `cv show <agent-id>` with no parent given: find which session(s) spawned the agent (filename
/// scan across the fleet) and render it through the one parent — or list the candidates when the
/// prefix is ambiguous. Returns `false` when nothing agent-shaped matched (the caller has one
/// more reading of the id to try).
fn show_agent_fleetwide(agent_id: &str, json: bool, window: &WindowArgs) -> Result<bool> {
    let parents = cv_core::find_subagent_parents(agent_id);
    // Copies of one agent across sessions (`cv adopt`) read through the newest parent's copy.
    let parents = if parents.len() > 1 && cv_core::one_agent_across(&parents, agent_id) {
        eprintln!(
            "✦ {} sessions hold a copy of this agent (adopted); reading the newest — `cv show <session> --agent {agent_id}` for another",
            parents.len()
        );
        parents[..1].to_vec()
    } else {
        parents
    };
    match parents.as_slice() {
        [] => Ok(false),
        [parent] => {
            let adapter = cv_core::harness::for_harness(parent.harness)
                .with_context(|| format!("no adapter for {}", parent.harness))?;
            eprintln!(
                "✦ sub-agent of session {} ({})",
                short_id(&parent.id),
                parent.title.as_deref().unwrap_or("untitled"),
            );
            show_one_subagent(parent, adapter.as_ref(), agent_id, json, window)?;
            Ok(true)
        }
        many => {
            // Ambiguous: every candidate as a pasteable `harness:id` line, exit 2.
            let mut lines: Vec<String> = many
                .iter()
                .map(|p| {
                    format!(
                        "{}:{}   # cv show {} --agent {agent_id} ({})",
                        p.harness.as_str(),
                        p.id,
                        short_id(&p.id),
                        p.title.as_deref().unwrap_or("untitled")
                    )
                })
                .collect();
            lines.sort();
            usage(format!(
                "ambiguous sub-agent id {agent_id:?} — {} sessions spawned a match:\n{}",
                many.len(),
                lines.join("\n")
            ))
        }
    }
}

/// List the sub-agent forest a session spawned (the `--subagents` view): every direct/`Workflow`
/// sub-agent with its type, journaled outcome, and final return value. `json` emits the structured
/// [`SubagentInfo`] forest (each annotated with its return) for machine consumption.
fn show_subagents(r: &SessionRef, json: bool) -> Result<()> {
    let subs = cv_core::subagent_tree_of(r);

    if json {
        // Enrich each with its final return value (direct agents report via the parent tool_result;
        // workflow agents via the journal summary already on the struct).
        let enriched: Vec<serde_json::Value> = subs
            .iter()
            .map(|s| {
                let mut v = serde_json::to_value(s).unwrap_or(serde_json::Value::Null);
                if let Some(obj) = v.as_object_mut() {
                    // Surface the bare agent id (the journal/transcript key) so consumers don't have
                    // to know the `agent-` stripping convention.
                    obj.insert("agent_id".into(), serde_json::Value::String(s.agent_id().to_string()));
                    // Direct agents have no journaled summary — attach their final return value.
                    if s.result_summary.is_none() {
                        if let Some(ret) = cv_core::harness::claude::subagent_return(&s.session.path) {
                            obj.insert("return".into(), serde_json::Value::String(ret));
                        }
                    }
                }
                v
            })
            .collect();
        println!("{}", serde_json::to_string_pretty(&enriched)?);
        return Ok(());
    }

    if subs.is_empty() {
        println!("no sub-agents spawned by {}", short_id(&r.id));
        return Ok(());
    }

    println!("# sub-agents of {}\n", short_id(&r.id));
    for s in &subs {
        let wf = s.workflow.as_deref().map(|w| format!("  ⟐{w}")).unwrap_or_default();
        println!("• {}{wf}", subagent_line(s));
        // A direct agent has no journaled summary (the line showed its task): add its real return,
        // the last text turn of its transcript. Transcript-derived, so sanitized at the seam (G5);
        // the `--json` path stays raw.
        if s.result_summary.is_none() {
            if let Some(ret) = cv_core::harness::claude::subagent_return(&s.session.path) {
                println!("  ↩ {}", truncate(&cv_core::sanitize::sanitize_line(&ret), 400));
            }
        }
    }
    Ok(())
}

/// One sub-agent of `parent` by its `agent-…` id, bare agentId, or a unique prefix of either — the
/// resolution `cv show --agent` and `cv rewind --agent` share. No match / several → exit 2.
pub(crate) fn resolve_subagent(parent: &SessionRef, agent_id: &str) -> Result<cv_core::SubagentInfo> {
    let subs = cv_core::subagent_tree_of(parent);
    // Match on the full session id (`agent-…`), the bare agentId, or a prefix of either.
    let matches: Vec<&cv_core::SubagentInfo> = subs
        .iter()
        .filter(|s| {
            s.session.id == agent_id
                || s.agent_id() == agent_id
                || s.session.id.starts_with(agent_id)
                || s.agent_id().starts_with(agent_id)
        })
        .collect();
    match matches.as_slice() {
        [one] => Ok((*one).clone()),
        [] => usage(format!(
            "no sub-agent matching {agent_id:?} under {} ({} sub-agent(s); try `cv show {} --subagents`)",
            short_id(&parent.id),
            subs.len(),
            short_id(&parent.id),
        )),
        many => {
            let mut lines: Vec<String> = many
                .iter()
                .map(|s| format!("{}:{}", s.session.harness.as_str(), s.session.id))
                .collect();
            lines.sort();
            usage(format!(
                "ambiguous sub-agent id {agent_id:?} — {} candidates under {}:\n{}",
                many.len(),
                short_id(&parent.id),
                lines.join("\n")
            ))
        }
    }
}

/// Render one specific sub-agent's transcript (`--agent <id>`), resolved by id-prefix relative to
/// its parent session. Honors `--json` and the window flags exactly as a top-level `cv show` would
/// (`--last` counts the sub-agent's own messages).
fn show_one_subagent(
    parent: &SessionRef,
    adapter: &dyn Adapter,
    agent_id: &str,
    json: bool,
    window: &WindowArgs,
) -> Result<()> {
    let sub = resolve_subagent(parent, agent_id)?;
    let range = window.window()?;

    if json {
        return print_session_json(adapter, &sub.session, range, window.max_bytes, false, None).map(|_| ());
    }

    // A small provenance banner so the reader knows which agent (and outcome) this is.
    if let Some(d) = &sub.description {
        println!("# sub-agent: {}", truncate(d, 120));
    }
    let wf = sub
        .workflow
        .as_deref()
        .map(|w| format!(" · workflow {w}"))
        .unwrap_or_default();
    let status = sub
        .result_status
        .as_deref()
        .map(|st| format!(" · {st}"))
        .unwrap_or_default();
    println!(
        "{} · {}{}{}\n",
        sub.agent_type.as_deref().unwrap_or("agent"),
        sub.agent_id(),
        status,
        wf,
    );
    render_text(adapter, &sub.session, range, window.max_bytes, false)
}

pub(crate) fn cmd_export(
    id: &str,
    format: &str,
    harness: Option<String>,
    window: &WindowArgs,
    redact: bool,
) -> Result<()> {
    use cv_core::redact::{redact_message, redact_text, RedactOptions, RedactStats};
    let want = parse_harness(&harness)?;
    let (r, adapter) = resolve(id, want)?;
    let range = window.window()?;
    // `--redact`: the same scrubber `cv share` runs (API keys, private keys, JWTs, emails,
    // assignments, high-entropy blobs), applied per block before anything is rendered; the
    // per-class counts go to stderr so stdout stays the document.
    let opts = RedactOptions::default();
    let stats = std::cell::RefCell::new(RedactStats::default());
    match format {
        "md" | "markdown" => {
            let mut out = std::io::BufWriter::new(std::io::stdout().lock());
            let outcome = stream_session_render(
                adapter.as_ref(),
                &r,
                &mut out,
                |h| {
                    let head = md_header(h);
                    if redact {
                        redact_text(&head)
                    } else {
                        head
                    }
                },
                |m| {
                    if !redact {
                        return md_message(m);
                    }
                    let mut m = m.clone();
                    redact_message(&mut m, &opts, &mut stats.borrow_mut());
                    md_message(&m)
                },
                RenderOpts {
                    range,
                    max_bytes: window.max_bytes,
                },
            )?;
            if let Some(next) = outcome.next {
                writeln!(out, "{}", continue_hint(next))?;
            }
            out.flush()?;
        }
        // `--format json` IS OpenSession 0.3: the IR, plus the version marker.
        "json" => {
            let st = print_session_json(
                adapter.as_ref(),
                &r,
                range,
                window.max_bytes,
                true,
                redact.then_some(&opts),
            )?;
            *stats.borrow_mut() = st;
        }
        "html" => {
            return usage(
                "`cv export --format html` was removed in 0.15.0 — `cv share <id> [--out FILE]` writes the \
                 self-contained, redacted HTML artifact (`--no-redact` to keep secrets)",
            )
        }
        other => bail!("unknown format {other:?} (use md or json; `cv share` for html)"),
    }
    if redact {
        eprintln!("{}", redaction_summary(&stats.borrow()));
    }
    Ok(())
}

/// The one-line per-class tally `export --redact` and `share` print to stderr.
pub(crate) fn redaction_summary(st: &cv_core::redact::RedactStats) -> String {
    format!(
        "✦ redacted {} item(s): {} api_key, {} private_key, {} jwt, {} email, {} blob, {} assignment",
        st.total(),
        st.api_keys,
        st.private_keys,
        st.jwts,
        st.emails,
        st.blobs,
        st.assignments,
    )
}

// ---------- show --tree ----------

fn print_tree(r: &SessionRef, adapter: &dyn Adapter) -> Result<()> {
    let session = adapter.parse(r)?;

    println!("# {}", session.label());
    println!("{} · {} · {} msg", session.harness, session.id, session.messages.len());
    println!();

    // Threaded view only if at least one message carries a parent_id.
    let has_threading = session.messages.iter().any(|m| m.parent_id.is_some());
    if has_threading {
        render_tree_dag(&session);
    } else {
        for (i, m) in session.messages.iter().enumerate() {
            println!("{:>4}. {}", i + 1, tree_line(m));
        }
    }

    // The second dimension: the sub-agent forest this session spawned. Claude sessions fan out
    // into directly-spawned (`Agent`/`Task`) sub-agents and `Workflow` sub-agents (grouped by run
    // id, with the orchestrator's journaled outcome) — invisible to flat message threading.
    render_subagent_forest(r);
    Ok(())
}

/// Print the sub-agent forest under a session: directly-spawned sub-agents, then each workflow's
/// agents grouped together with the journaled `status: summary` outcome. Reads only the cheap
/// metadata/journal sidecars (no transcript bodies).
fn render_subagent_forest(r: &SessionRef) {
    use std::collections::BTreeMap;
    let subs = cv_core::subagent_tree_of(r);
    if subs.is_empty() {
        return;
    }

    let direct: Vec<&cv_core::SubagentInfo> = subs.iter().filter(|s| s.workflow.is_none()).collect();
    // Workflow agents grouped by run id (BTreeMap → stable run order).
    let mut by_wf: BTreeMap<&str, Vec<&cv_core::SubagentInfo>> = BTreeMap::new();
    for s in subs.iter().filter(|s| s.workflow.is_some()) {
        by_wf.entry(s.workflow.as_deref().unwrap_or("")).or_default().push(s);
    }

    println!();
    println!(
        "## sub-agents ({} direct, {} workflow agent(s) across {} workflow(s))",
        direct.len(),
        subs.len() - direct.len(),
        by_wf.len()
    );

    if !direct.is_empty() {
        println!();
        for s in &direct {
            println!("• {}", subagent_line(s));
        }
    }
    for (wf, agents) in &by_wf {
        println!("\n  ⟐ workflow {wf}  ({} agent(s))", agents.len());
        for s in agents {
            println!("    • {}", subagent_line(s));
        }
    }
}

/// One line describing a sub-agent — `cv show --tree`'s forest and `cv show --subagents` share it: short
/// id, type, the journaled status (workflow agents), message count, and the journaled summary or,
/// failing that, the task description — capped and sanitized (both are transcript-derived).
fn subagent_line(s: &cv_core::SubagentInfo) -> String {
    let id = short_id(s.agent_id());
    let kind = s.agent_type.as_deref().unwrap_or("agent");
    let status = s
        .result_status
        .as_deref()
        .map(|st| format!(" [{st}]"))
        .unwrap_or_default();
    // Prefer the journaled summary (the real outcome); fall back to the task description.
    let blurb = s
        .result_summary
        .as_deref()
        .or(s.description.as_deref())
        .map(|t| truncate(&cv_core::sanitize::sanitize_line(t), 88))
        .unwrap_or_default();
    format!("{id}  {kind}{status}  {} msg  {blurb}", s.session.message_count)
}

/// Render messages as an indented DAG by `parent_id`. Roots (no/unknown parent) sit at depth 0.
fn render_tree_dag(session: &Session) {
    use std::collections::HashMap;
    // children: parent_id -> ordered list of child message indices.
    let mut by_id: HashMap<&str, usize> = HashMap::new();
    for (i, m) in session.messages.iter().enumerate() {
        if let Some(id) = &m.id {
            by_id.insert(id.as_str(), i);
        }
    }
    let mut children: HashMap<Option<usize>, Vec<usize>> = HashMap::new();
    for (i, m) in session.messages.iter().enumerate() {
        let parent = m
            .parent_id
            .as_deref()
            .and_then(|p| by_id.get(p).copied())
            .filter(|&p| p != i);
        children.entry(parent).or_default().push(i);
    }

    fn walk(
        node: Option<usize>,
        depth: usize,
        session: &Session,
        children: &std::collections::HashMap<Option<usize>, Vec<usize>>,
    ) {
        if let Some(kids) = children.get(&node) {
            for &c in kids {
                let indent = "  ".repeat(depth);
                println!("{indent}• {}", tree_line(&session.messages[c]));
                walk(Some(c), depth + 1, session, children);
            }
        }
    }
    walk(None, 0, session, &children);
}

/// One-line preview of a message for the tree: role (with its kind when that says more than the
/// role), markers for tool turns / sub-agent spawns, and a text preview.
fn tree_line(m: &Message) -> String {
    let role = turn_label(m);
    let mut tags = Vec::new();
    let has_tool_use = m.content.iter().any(|b| matches!(b, Block::ToolUse { .. }));
    let has_tool_result = m.content.iter().any(|b| matches!(b, Block::ToolResult { .. }));
    if has_tool_use {
        tags.push("🔧 tool".to_string());
        // Surface a sub-agent spawn if the tool looks like one.
        for b in &m.content {
            if let Block::ToolUse { name, .. } = b {
                let n = name.to_ascii_lowercase();
                if n.contains("task") || n.contains("agent") || n.contains("dispatch") || n.contains("spawn") {
                    tags.push(format!("↳ sub-agent ({name})"));
                }
            }
        }
    }
    if has_tool_result {
        tags.push("↩ result".to_string());
    }
    // The adapter said so: a spawn the tool-name heuristic didn't catch.
    if m.kind == MessageKind::SubagentSpawn && !tags.iter().any(|t| t.starts_with("↳ sub-agent")) {
        tags.push("↳ sub-agent".to_string());
    }

    let preview = m.text().map(|t| truncate(&t, 80)).unwrap_or_else(|| {
        if has_tool_use {
            m.content
                .iter()
                .find_map(|b| match b {
                    Block::ToolUse { name, .. } => Some(format!("[{name}]")),
                    _ => None,
                })
                .unwrap_or_default()
        } else {
            String::new()
        }
    });

    let tagstr = if tags.is_empty() {
        String::new()
    } else {
        format!(" [{}]", tags.join(", "))
    };
    format!("{role}{tagstr}  {preview}")
}

// ---------- rendering helpers ----------

/// Everything a streamed renderer needs for a session's header — derived from the cheap
/// [`SessionRef`] (label/cwd) plus the model captured from the first assistant turn.
pub(crate) struct HeaderInfo {
    harness: Harness,
    id: String,
    cwd: Option<PathBuf>,
    label: String,
    model: Option<String>,
    created_at: Option<chrono::DateTime<chrono::Utc>>,
}

/// What to render: an optional message window and an optional byte budget.
#[derive(Debug, Clone, Copy, Default)]
pub(crate) struct RenderOpts {
    /// `Last(n)` is resolved in the render's own pass: it keeps the last `n` messages as they
    /// stream by and renders them at the end, so `--last` reads the session once, not twice.
    pub range: Option<Window>,
    /// Stop before the message that would push the rendered output past this many bytes (the
    /// first in-window message is always rendered, so progress is always possible).
    pub max_bytes: Option<usize>,
}

/// What a render did: how many messages it wrote, and — when the byte budget cut it short — the
/// index of the first message it did NOT write (the `--range <next>..` continuation point).
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub(crate) struct RenderOutcome {
    pub rendered: usize,
    pub next: Option<usize>,
}

/// Render a session to `out` by **streaming** — each message is rendered and written as it arrives,
/// then dropped, so a multi-GB transcript renders at O(largest message) instead of materializing the
/// whole `Session`. The header needs the label and model, which come from the first user/assistant
/// turns, so it's held behind a small bounded buffer (`HOLDBACK` messages) until those are known and
/// then flushed ahead of the body — header info lives in the first turns, so the buffer stays tiny.
pub(crate) fn stream_session_render<W: std::io::Write>(
    adapter: &dyn Adapter,
    r: &SessionRef,
    out: &mut W,
    header: impl Fn(&HeaderInfo) -> String,
    render_msg: impl Fn(&Message) -> String,
    opts: RenderOpts,
) -> Result<RenderOutcome> {
    use cv_core::{Flow, MessageSink, ParseOptions};
    const HOLDBACK: usize = 24;

    /// The header fields a `meta` call delivers.
    struct Meta {
        title: Option<String>,
        model: Option<String>,
        cwd: Option<PathBuf>,
        created_at: Option<chrono::DateTime<chrono::Utc>>,
    }

    /// `--last n` in one pass: the last `n` messages, still lazy, plus what the render would have
    /// learned on the way — each `meta` with the message count it arrived after, and the first
    /// message model — so [`Sink::finish_tail`] replays the window exactly as a second pass that
    /// started at `total - n` would have.
    struct Tail {
        n: usize,
        ring: std::collections::VecDeque<Message>,
        seen: usize,
        metas: Vec<(usize, Meta)>,
        first_model: Option<(usize, String)>,
    }

    struct Sink<'w, W, H, R> {
        out: &'w mut W,
        harness: Harness,
        id: String,
        cwd: Option<PathBuf>,
        title: Option<String>,
        model: Option<String>,
        created_at: Option<chrono::DateTime<chrono::Utc>>,
        first_user: Option<String>,
        header: H,
        render_msg: R,
        buf: Vec<String>,
        printed: bool,
        result: std::io::Result<()>,
        resolver: cv_core::Resolver,
        // Windowed view: 0-based message index we're at, plus the [start, end) bounds. Messages
        // outside the window are never materialized — their on-disk content is never touched.
        idx: usize,
        start: usize,
        end: Option<usize>,
        // Byte budget: rendered bytes so far (header + messages) and where the cut landed.
        max_bytes: Option<usize>,
        bytes: usize,
        rendered: usize,
        next: Option<usize>,
        tail: Option<Tail>,
    }
    impl<W: std::io::Write, H: Fn(&HeaderInfo) -> String, R: Fn(&Message) -> String> Sink<'_, W, H, R> {
        fn apply_meta(&mut self, s: Meta) {
            self.title = s.title;
            if self.model.is_none() {
                self.model = s.model;
            }
            if self.cwd.is_none() {
                self.cwd = s.cwd;
            }
            if self.created_at.is_none() {
                self.created_at = s.created_at;
            }
        }
        /// Render the kept tail as the second pass would have: everything before the window
        /// (metas, the first message model) applied in stream order, then the window's messages
        /// with each meta delivered before the message it preceded.
        fn finish_tail(&mut self) {
            let Some(t) = self.tail.take() else { return };
            let start = t.seen - t.ring.len();
            self.start = start;
            self.idx = start;
            let mut metas = t.metas.into_iter().peekable();
            let mut pre_model = t.first_model.filter(|(i, _)| *i < start);
            while let Some((at, m)) = metas.next_if(|(at, _)| *at <= start) {
                if let Some((_, md)) = pre_model.take_if(|(i, _)| *i < at) {
                    self.model.get_or_insert(md);
                }
                self.apply_meta(m);
            }
            if let Some((_, md)) = pre_model {
                self.model.get_or_insert(md);
            }
            for (k, m) in t.ring.into_iter().enumerate() {
                while let Some((_, s)) = metas.next_if(|(at, _)| *at <= start + k) {
                    self.apply_meta(s);
                }
                if self.message(m) == Flow::Stop {
                    break;
                }
            }
            for (_, s) in metas {
                self.apply_meta(s);
            }
        }
        fn write(&mut self, s: &str) {
            if self.result.is_ok() {
                self.result = self.out.write_all(s.as_bytes());
            }
        }
        fn flush_header(&mut self) {
            if self.printed {
                return;
            }
            let info = HeaderInfo {
                harness: self.harness,
                id: self.id.clone(),
                cwd: self.cwd.clone(),
                label: cv_core::label_from(self.title.as_deref(), self.first_user.as_deref()),
                model: self.model.clone(),
                created_at: self.created_at,
            };
            let head = (self.header)(&info);
            self.bytes += head.len();
            self.write(&head);
            let buffered = std::mem::take(&mut self.buf);
            for s in buffered {
                self.write(&s);
            }
            self.printed = true;
        }
    }
    impl<W: std::io::Write, H: Fn(&HeaderInfo) -> String, R: Fn(&Message) -> String> MessageSink for Sink<'_, W, H, R> {
        fn meta(&mut self, s: &Session) {
            // Authoritative *parsed* session metadata, delivered before the body by the bridge
            // `stream` and by adapters that call `sink.meta` (e.g. codex). The parsed title overrides
            // the discovery-time `SessionRef` title so the header matches a full parse — some
            // adapters' discovery title differs from the parsed one (codex's discovery title is the
            // first user record, which the parse skips). Model/cwd fill in if discovery lacked them.
            let meta = Meta {
                title: s.title.clone(),
                model: s.model.clone(),
                cwd: s.cwd.clone(),
                created_at: s.created_at,
            };
            match self.tail.as_mut() {
                Some(t) => t.metas.push((t.seen, meta)),
                None => self.apply_meta(meta),
            }
        }
        fn message(&mut self, m: Message) -> Flow {
            if let Some(t) = self.tail.as_mut() {
                if t.first_model.is_none() {
                    if let Some(md) = &m.model {
                        t.first_model = Some((t.seen, md.clone()));
                    }
                }
                t.seen += 1;
                if t.n > 0 {
                    t.ring.push_back(m);
                    if t.ring.len() > t.n {
                        t.ring.pop_front();
                    }
                }
                return Flow::Continue;
            }
            if self.result.is_err() || self.next.is_some() {
                return Flow::Stop;
            }
            let idx = self.idx;
            self.idx += 1;

            // Past the window's end: nothing left to render — stop early so we never read the
            // rest of the file (the whole point of a windowed show on a huge session).
            if let Some(end) = self.end {
                if idx >= end {
                    return Flow::Stop;
                }
            }

            // Model is a small already-parsed field — pick it up even from out-of-window messages
            // so the header stays accurate, without touching any content bytes.
            if self.model.is_none() {
                if let Some(md) = &m.model {
                    self.model = Some(md.clone());
                }
            }

            // Before the window: skip entirely. We do NOT materialize, so no span content is read.
            if idx < self.start {
                return Flow::Continue;
            }

            // In-window: resolve this message's lazy content spans (peak = one message) and render.
            let mut m = m;
            m.materialize(&self.resolver);
            if self.first_user.is_none() && m.role == Role::User {
                if let Some(t) = m.text() {
                    if !t.trim().is_empty() {
                        self.first_user = Some(t);
                    }
                }
            }
            let rendered = (self.render_msg)(&m);
            drop(m);
            // The byte budget: stop BEFORE the message that would overflow it — unless it is the
            // first in-window message, which always renders so a window never comes back empty.
            if let Some(budget) = self.max_bytes {
                if self.rendered > 0 && self.bytes + rendered.len() > budget {
                    self.next = Some(idx);
                    return Flow::Stop;
                }
            }
            self.bytes += rendered.len();
            self.rendered += 1;
            if self.printed {
                self.write(&rendered);
            } else {
                self.buf.push(rendered);
                let title_known = self.title.is_some() || self.first_user.is_some();
                if (self.model.is_some() && title_known) || self.buf.len() >= HOLDBACK {
                    self.flush_header();
                }
            }
            if self.result.is_err() {
                Flow::Stop
            } else {
                Flow::Continue
            }
        }
    }

    let (start, end, tail) = match opts.range {
        None => (0, None, None),
        Some(Window::Last(n)) => (0, None, Some(n)),
        Some(w) => {
            let (s, e) = w.bounds(|| unreachable!("only `Last` needs the total, and it streams instead"))?;
            (s, e, None)
        }
    };
    let mut sink = Sink {
        out,
        harness: r.harness,
        id: r.id.clone(),
        cwd: r.cwd.clone(),
        title: r.title.clone(),
        model: None,
        created_at: r.created_at,
        first_user: None,
        header,
        render_msg,
        buf: Vec::new(),
        printed: false,
        result: Ok(()),
        resolver: cv_core::Resolver::new(Some(r.path.clone())),
        idx: 0,
        start,
        end,
        max_bytes: opts.max_bytes,
        bytes: 0,
        rendered: 0,
        next: None,
        tail: tail.map(|n| Tail {
            n,
            ring: Default::default(),
            seen: 0,
            metas: Vec::new(),
            first_model: None,
        }),
    };
    // Windowed: lazy spans, so out-of-window giant fields arrive as 16-byte handles and only
    // in-window messages materialize (the sink already resolves per message). A full show reads
    // every byte regardless — spans would only add resolve overhead there (floor = C).
    let parse_opts = if opts.range.is_some() {
        ParseOptions::lazy()
    } else {
        ParseOptions::bulk()
    };
    cv_core::stream::stream_counted(adapter, r, &parse_opts, &mut sink)?;
    sink.finish_tail();
    sink.flush_header(); // short sessions (no assistant turn / < HOLDBACK msgs) flush here
    sink.result?;
    Ok(RenderOutcome {
        rendered: sink.rendered,
        next: sink.next,
    })
}

/// Header for `cv show` (mirrors the old eager header exactly).
pub(crate) fn show_header(h: &HeaderInfo) -> String {
    format!(
        "# {}\n{} · {} · {}{}\n\n",
        h.label,
        h.harness,
        h.id,
        h.cwd.as_deref().map(home_rel).unwrap_or_else(|| "?".into()),
        h.model.as_ref().map(|m| format!(" · {m}")).unwrap_or_default(),
    )
}

/// One rendered `cv show` message block (the String form of the old `print_message`).
pub(crate) fn show_message(m: &Message) -> String {
    let mut s = format!("── {} ──\n", turn_label(m));
    for b in &m.content {
        match b {
            Block::Text { text } => {
                s.push_str(text);
                s.push('\n');
            }
            Block::Thinking { text, .. } => s.push_str(&format!("[thinking] {}\n", truncate(text, 200))),
            Block::ToolUse { id, name, input, .. } => s.push_str(&format!(
                "[tool_use {name} {id}] {}\n",
                truncate(&input.to_string(), 200)
            )),
            Block::ToolResult {
                tool_use_id,
                content,
                is_error,
                details,
                ..
            } => {
                s.push_str(&format!(
                    "[tool_result{} {tool_use_id}] {}\n",
                    if *is_error { " error" } else { "" },
                    truncate(content, 200)
                ));
                if let Some(p) = persisted_path(details.as_ref()) {
                    s.push_str(&format!(
                        "  ↳ full output on disk: {p} (cv cat <session> {tool_use_id})\n"
                    ));
                }
            }
            Block::File { path, source, .. } => s.push_str(&format!(
                "[file: {}]\n",
                path.as_deref().or(source.as_deref()).unwrap_or("?")
            )),
            Block::Image { .. } => s.push_str("[image]\n"),
        }
    }
    s.push('\n');
    s
}

/// Where a persisted (too-large) tool output lives on disk, when the adapter recorded it — the
/// transcript holds only the stub the model saw.
fn persisted_path(details: Option<&serde_json::Value>) -> Option<&str> {
    details?.pointer("/persistedOutput/path")?.as_str()
}

/// Header for `cv export md`: [`cv_core::render::md_header`] from the streamed header facts.
fn md_header(h: &HeaderInfo) -> String {
    cv_core::render::md_header(
        &h.label,
        h.harness,
        &h.id,
        h.cwd.as_deref(),
        h.model.as_deref(),
        h.created_at,
    )
}

/// One `cv export md` section: [`cv_core::render::md_message`] as a `String`.
fn md_message(m: &Message) -> String {
    let mut out = String::new();
    cv_core::render::md_message(m, &mut out);
    out
}

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

    /// `--kind` picks one view family: what the person said, the compaction seams, or the
    /// affordance rows (`affordances` meaning all of them); mixing families or an unknown word is a
    /// usage error naming every accepted word.
    #[test]
    fn kind_words_pick_one_view_family() {
        use cv_core::affordance::Kind;
        assert!(matches!(
            parse_kinds("prompt,answer"),
            Ok(KindView::Said {
                prompts: true,
                answers: true
            })
        ));
        assert!(matches!(parse_kinds("compaction_summary"), Ok(KindView::Compactions)));
        match parse_kinds("plan,schedule") {
            Ok(KindView::Affordances(k)) => assert_eq!(k, [Kind::Plan, Kind::Schedule].into_iter().collect()),
            _ => panic!("plan,schedule is the affordance view"),
        }
        match parse_kinds("affordances") {
            Ok(KindView::Affordances(k)) => assert_eq!(k.len(), Kind::ALL.len()),
            _ => panic!("affordances is every kind"),
        }
        let err = parse_kinds("plan,prompt").err().expect("mixed families").to_string();
        assert!(err.contains("mixes two views"), "{err}");
        let err = parse_kinds("todo").err().expect("unknown word").to_string();
        assert!(err.contains("background_job"), "{err}");
    }

    fn temp_home(tag: &str) -> PathBuf {
        let d = std::env::temp_dir().join(format!("cv-view-render-{tag}-{}", std::process::id()));
        std::fs::create_dir_all(&d).unwrap();
        d
    }

    fn render(r: &SessionRef, range: Option<(usize, Option<usize>)>) -> String {
        let range = range.map(|(a, b)| Window::Range(a, b));
        render_with(r, RenderOpts { range, max_bytes: None }).0
    }

    fn render_with(r: &SessionRef, opts: RenderOpts) -> (String, RenderOutcome) {
        let adapter = cv_core::harness::for_harness(r.harness).unwrap();
        let mut out = Vec::new();
        let outcome = stream_session_render(adapter.as_ref(), r, &mut out, show_header, show_message, opts).unwrap();
        (String::from_utf8(out).unwrap(), outcome)
    }

    /// Render the window `range` of `r` and check its body is each message of a full eager parse,
    /// rendered alone and concatenated. Returns the windowed render's header.
    fn assert_window_matches_full_parse(r: &SessionRef, range: (usize, Option<usize>)) -> String {
        let adapter = cv_core::harness::for_harness(r.harness).unwrap();
        let mut all = cv_core::CollectSink::default();
        adapter.stream(r, &cv_core::ParseOptions::bulk(), &mut all).unwrap();
        let end = range.1.unwrap_or(usize::MAX).min(all.messages.len());
        let want: String = all.messages[range.0..end].iter().map(show_message).collect();
        let windowed = render(r, Some(range));
        let body_at = windowed.find("── ").unwrap_or(windowed.len());
        assert_eq!(
            &windowed[body_at..],
            want,
            "range {range:?} body must equal the full parse's slice"
        );
        windowed[..body_at].to_string()
    }

    /// A small Claude fixture: title, a user turn, a model-carrying assistant turn, one big user
    /// turn, then eight short follow-ups (11 messages).
    fn claude_fixture(home: &std::path::Path) -> SessionRef {
        let path = home.join("s.jsonl");
        let big = "long \"quoted\" content\n".repeat(400);
        let mut f = std::fs::File::create(&path).unwrap();
        writeln!(f, r#"{{"type":"ai-title","aiTitle":"render window"}}"#).unwrap();
        writeln!(
            f,
            r#"{{"type":"user","uuid":"u0","cwd":"/w","message":{{"role":"user","content":"q one"}}}}"#
        )
        .unwrap();
        writeln!(
            f,
            "{}",
            serde_json::json!({"type":"assistant","uuid":"a1","message":{"role":"assistant",
                "model":"claude-test","content":[{"type":"text","text":"a one"}]}})
        )
        .unwrap();
        writeln!(
            f,
            "{}",
            serde_json::json!({"type":"user","uuid":"u2","message":{"role":"user","content":big}})
        )
        .unwrap();
        for i in 0..8 {
            writeln!(
                f,
                "{}",
                serde_json::json!({"type":"user","uuid":format!("u{}", 3+i),
                    "message":{"role":"user","content":format!("follow-up {i}")}})
            )
            .unwrap();
        }
        drop(f);
        SessionRef {
            id: "render-window".into(),
            harness: Harness::Claude,
            path,
            cwd: Some("/w".into()),
            title: Some("render window".into()),
            created_at: None,
            updated_at: None,
            message_count: 11,
        }
    }

    /// The windowed-render contract: a window renders byte-identically to the same messages of a
    /// full parse, and its header carries the model the skipped prefix provides.
    #[test]
    fn windowed_render_is_byte_identical_to_the_full_parse() {
        let home = temp_home("claude");
        let r = claude_fixture(&home);
        let full_header = {
            let full = render(&r, None);
            full[..full.find("── ").unwrap()].to_string()
        };
        for range in [(3usize, Some(6usize)), (2, Some(11)), (5, None), (1, Some(2))] {
            let header = assert_window_matches_full_parse(&r, range);
            assert_eq!(header, full_header, "range {range:?} header");
            assert!(header.contains("claude-test"), "header model expected in {range:?}");
        }
        std::fs::remove_dir_all(&home).ok();
    }

    /// `--max-bytes`: the render stops before the message that would overflow the budget and
    /// reports the continuation index; the first in-window message always renders; a budget
    /// that fits everything reports no cut.
    #[test]
    fn byte_budget_cuts_and_reports_the_next_index() {
        let home = temp_home("budget");
        let r = claude_fixture(&home);

        // Message 2 is the ~9 KB giant: a 1 KB budget renders messages 0 and 1, then stops.
        let (out, oc) = render_with(
            &r,
            RenderOpts {
                range: None,
                max_bytes: Some(1024),
            },
        );
        assert_eq!(oc.rendered, 2, "{out}");
        assert_eq!(oc.next, Some(2), "{out}");
        assert!(
            out.contains("q one") && out.contains("a one") && !out.contains("follow-up"),
            "{out}"
        );

        // Starting AT the giant: it renders anyway (progress is always possible), then the cut.
        let (out, oc) = render_with(
            &r,
            RenderOpts {
                range: Some(Window::Range(2, None)),
                max_bytes: Some(1024),
            },
        );
        assert_eq!(oc.rendered, 1, "{out}");
        assert_eq!(oc.next, Some(3), "{out}");
        assert!(out.contains("long \"quoted\" content"), "{out}");

        // A generous budget: everything, no cut.
        let (_, oc) = render_with(
            &r,
            RenderOpts {
                range: Some(Window::Range(3, None)),
                max_bytes: Some(1 << 20),
            },
        );
        assert_eq!(oc.rendered, 8);
        assert_eq!(oc.next, None);

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

    /// `--last n` keeps a ring of the last `n` messages in one pass; it must render exactly what the
    /// counted `total - n ..` range renders, budget cut and continuation index included.
    #[test]
    fn last_n_renders_what_the_counted_range_does() {
        let home = temp_home("last");
        let r = claude_fixture(&home);
        for n in [0usize, 1, 3, 10, 11, 50] {
            for max_bytes in [None, Some(1024)] {
                let start = 11usize.saturating_sub(n);
                let counted = render_with(
                    &r,
                    RenderOpts {
                        range: Some(Window::Range(start, None)),
                        max_bytes,
                    },
                );
                let ring = render_with(
                    &r,
                    RenderOpts {
                        range: Some(Window::Last(n)),
                        max_bytes,
                    },
                );
                assert_eq!(ring, counted, "--last {n} (budget {max_bytes:?})");
            }
        }
        std::fs::remove_dir_all(&home).ok();
    }

    /// A store whose metadata arrives after the body (codex delivers `meta` at EOF when no model is
    /// ever known) and whose model first shows up past `HOLDBACK` messages: the ring replays each
    /// `meta` and the prefix's model where a second pass would have met them, so the header is the
    /// counted render's, byte for byte.
    #[test]
    fn last_n_replays_late_meta_and_models_in_stream_order() {
        struct Late(usize);
        impl Adapter for Late {
            fn harness(&self) -> Harness {
                Harness::Codex
            }
            fn roots(&self) -> Vec<PathBuf> {
                Vec::new()
            }
            fn discover(&self) -> Result<Vec<SessionRef>> {
                Ok(Vec::new())
            }
            fn stream(
                &self,
                r: &SessionRef,
                _: &cv_core::ParseOptions,
                sink: &mut dyn cv_core::MessageSink,
            ) -> Result<Session> {
                let mut s = Session::new(r.id.clone(), Harness::Codex);
                for i in 0..self.0 {
                    if i == 12 {
                        s.title = Some("mid-stream title".into());
                        sink.meta(&s);
                    }
                    let mut m = Message::new(if i % 2 == 0 { Role::User } else { Role::Assistant });
                    if i == 30 || i == 45 {
                        m.model = Some(format!("model-{i}"));
                    }
                    m.content.push(Block::Text {
                        text: format!("turn {i}").into(),
                    });
                    if sink.message(m) == cv_core::Flow::Stop {
                        break;
                    }
                }
                s.title = Some("eof title".into());
                s.cwd = Some("/late".into());
                sink.meta(&s);
                Ok(s)
            }
        }
        let r = SessionRef {
            id: "late".into(),
            harness: Harness::Codex,
            path: "/nonexistent/late.jsonl".into(),
            cwd: None,
            title: Some("discovery title".into()),
            created_at: None,
            updated_at: None,
            message_count: 0,
        };
        let total = 60usize;
        let a = Late(total);
        let render = |range: Window, max_bytes: Option<usize>| {
            let mut out = Vec::new();
            let oc = stream_session_render(
                &a,
                &r,
                &mut out,
                show_header,
                show_message,
                RenderOpts {
                    range: Some(range),
                    max_bytes,
                },
            )
            .unwrap();
            (String::from_utf8(out).unwrap(), oc)
        };
        for n in [0usize, 5, 20, 29, 30, 31, 40, 48, 60, 100] {
            for max_bytes in [None, Some(200)] {
                let counted = render(Window::Range(total.saturating_sub(n), None), max_bytes);
                assert_eq!(
                    render(Window::Last(n), max_bytes),
                    counted,
                    "--last {n} (budget {max_bytes:?})"
                );
            }
        }
    }

    /// One label spelling and one block text on every door: `cv show`, `cv export --format md`
    /// and `cv share` label a turn with `render::turn_label` and carry its text blocks verbatim.
    #[test]
    fn show_export_and_share_agree_on_labels_and_block_text() {
        let mut sys = Message::new(Role::System);
        sys.kind = MessageKind::InjectedContext;
        sys.harness_extra_mut(Harness::Claude)
            .insert("attachment_type".into(), serde_json::json!("hook_success"));
        sys.content.push(Block::Text {
            text: "reminder: run the gate".into(),
        });
        let mut summary = Message::new(Role::User);
        summary.kind = MessageKind::CompactionSummary;
        summary.content.push(Block::Text {
            text: "earlier we fixed the parser".into(),
        });
        let mut reply = Message::new(Role::Assistant);
        reply.content.push(Block::Text {
            text: "on it, editing ir.rs".into(),
        });
        for (m, label) in [
            (&sys, "system · injected_context/hook_success"),
            (&summary, "user · compaction_summary"),
            (&reply, "assistant"),
        ] {
            assert_eq!(turn_label(m), label);
            let text = m.text().unwrap();
            let shown = show_message(m);
            assert!(
                shown.starts_with(&format!("── {label} ──\n")) && shown.contains(&text),
                "{shown}"
            );
            let md = md_message(m);
            assert!(md.starts_with(&format!("## {label}\n\n")) && md.contains(&text), "{md}");
            let mut html = String::new();
            cv_core::html::share_message(m, 0, &mut html);
            assert!(
                html.contains(&format!("<span class=\"role\">{label}</span>")) && html.contains(&text),
                "{html}"
            );
        }
        // A flat legacy key is not an attachment kind: only a nested harness bag counts.
        let mut legacy = Message::new(Role::System);
        legacy.kind = MessageKind::Error;
        legacy.extra.insert("subtype".into(), serde_json::json!("api_error"));
        assert_eq!(turn_label(&legacy), "system · error");
    }

    /// Same contract for codex, whose metadata arrives through `meta()` (the parse-time title —
    /// `None` — overrides the discovery title).
    #[test]
    fn codex_windowed_render_is_byte_identical_to_the_full_parse() {
        let home = temp_home("codex");

        let path = home.join("rollout-r.jsonl");
        let big = "giant output\n".repeat(500);
        let mut f = std::fs::File::create(&path).unwrap();
        for l in [
            r#"{"timestamp":"2026-01-01T00:00:00Z","type":"session_meta","payload":{"id":"rollout-r","cwd":"/work"}}"#.to_string(),
            r#"{"timestamp":"2026-01-01T00:00:01Z","type":"turn_context","payload":{"cwd":"/work","model":"gpt-test"}}"#.to_string(),
            r#"{"timestamp":"2026-01-01T00:00:02Z","type":"event_msg","payload":{"type":"user_message","message":"hello there"}}"#.to_string(),
            r#"{"timestamp":"2026-01-01T00:00:03Z","type":"event_msg","payload":{"type":"agent_message","message":"working"}}"#.to_string(),
            r#"{"timestamp":"2026-01-01T00:00:04Z","type":"response_item","payload":{"type":"function_call","name":"shell","arguments":"{}","call_id":"c1"}}"#.to_string(),
            serde_json::json!({"timestamp":"2026-01-01T00:00:05Z","type":"response_item",
                "payload":{"type":"function_call_output","call_id":"c1","output":big}}).to_string(),
            r#"{"timestamp":"2026-01-01T00:00:06Z","type":"event_msg","payload":{"type":"agent_message","message":"all done"}}"#.to_string(),
        ] {
            writeln!(f, "{l}").unwrap();
        }
        drop(f);
        let r = SessionRef {
            id: "rollout-r".into(),
            harness: Harness::Codex,
            path: path.clone(),
            cwd: Some("/work".into()),
            title: Some("discovery title".into()),
            created_at: None,
            updated_at: None,
            message_count: 3,
        };
        for range in [(2usize, Some(4usize)), (1, None), (3, Some(5))] {
            let header = assert_window_matches_full_parse(&r, range);
            assert!(header.contains("gpt-test"), "header model expected in {range:?}");
            assert!(!header.contains("discovery title"), "parsed title wins in {range:?}");
        }

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