forge_agent/userdata/catalog/

scan.rs

//! Offline session scanners - stateless filesystem helpers that read
//! transcripts from `<config_dir>/projects/<project_key>/*.jsonl`.
//!
//! - [`list_sessions`] - lists sessions, either for one project or all.
//! - [`get_session_messages`] - reads the full transcript for one session.
//!
//! Session metadata ([`list_sessions`]) comes from an internal head +
//! tail lite read, so the fields it parses cost two 64 KiB reads
//! whatever the transcript's size. The worker tag is the exception and
//! is read in full: it can sit anywhere in the file and the last one
//! wins, so there is no window that settles it.

use std::collections::{HashMap, HashSet};
use std::fs;
use std::io::{BufRead, BufReader, Read, Seek, SeekFrom};
use std::path::{Path, PathBuf};

use serde_json::Value;
use unicode_normalization::UnicodeNormalization;
use uuid::Uuid;

use forge_primitives::{
    FORGE_WORKER_TAG_PREFIX, SDKSessionInfo, SessionHistory, SessionMessage, SessionMessageKind,
    worker_tag,
};
use forge_sdk::projects_dir_for;

/// True if `s` is a canonical 8-4-4-4-12 hyphenated UUID. The length
/// guard rejects the hyphenless / braced / URN forms that
/// `Uuid::try_parse` otherwise accepts - session ids on disk are always
/// the hyphenated form the CLI emits.
pub(crate) fn is_valid_uuid(s: &str) -> bool {
    s.len() == 36 && Uuid::try_parse(s).is_ok()
}

const MAX_SANITIZED_LENGTH: usize = 200;

/// Open `dir` for directory iteration. NotFound is the expected case
/// for the catalog's projects/ tree on a fresh forge install and is
/// silent; real I/O failures (perm denied, broken FS) log at warn
/// so the user gets a triage signal rather than silently empty
/// catalog reads.
fn try_read_dir(dir: &Path) -> Option<fs::ReadDir> {
    match fs::read_dir(dir) {
        Ok(iter) => Some(iter),
        Err(e) if e.kind() == std::io::ErrorKind::NotFound => None,
        Err(e) => {
            tracing::warn!(
                target: "forge_agent::userdata::catalog",
                path = %dir.display(),
                error = %e,
                "failed to read catalog directory"
            );
            None
        }
    }
}

/// Size of the head / tail byte buffer for lite metadata reads.
/// The CLI constant - match exactly so
/// the two implementations slice transcripts at the same boundary.
const LITE_READ_BUF_SIZE: u64 = 65_536;

/// Crate-internal re-export of the path sanitiser - other modules need
/// it to derive the same on-disk project-key layout the CLI uses. Not
/// part of the public API; downstream consumers should call
/// [`project_key_for_directory`] instead.
pub(crate) fn sanitize_path_public(name: &str) -> String {
    sanitize_path(name)
}

/// Map a directory path to the CLI's on-disk project key. Canonicalises
/// the path first and then applies the CLI's JS-style sanitisation
/// hash. `None` defaults to `"."` (the process's current working
/// directory).
pub fn project_key_for_directory(path: Option<&str>) -> String {
    sanitize_path(&canonicalize_path(path.unwrap_or(".")))
}

/// Resolve a directory to its realpath and apply NFC normalisation.
/// Wraps the CLI's `_canonicalize_path` (falls back to the input,
/// NFC-normalised, when the path can't be canonicalised: most
/// commonly because it doesn't exist). NFC is essential on
/// filesystems that don't auto-normalise (Linux ext4, Windows NTFS)
/// so decomposed inputs still hash to the CLI's on-disk
/// project-key layout.
fn canonicalize_path(path: &str) -> String {
    let resolved = match fs::canonicalize(path) {
        Ok(p) => p.to_string_lossy().into_owned(),
        Err(_) => path.to_string(),
    };
    resolved.nfc().collect()
}

/// Whether this read has already kept a row under `uuid`.
///
/// **The key a re-appended block is deduped by.** A transcript can carry a
/// block of records twice: the copies share `uuid`, `type` and the message
/// content, and may differ in metadata fields (`slug`, `promptId`, a zeroed
/// `usage`). `uuid` is what names the pair, and a row with no `uuid` names
/// nothing to dedupe on and is never a repeat.
fn already_kept(seen: &mut std::collections::HashSet<String>, uuid: &str) -> bool {
    !uuid.is_empty() && !seen.insert(uuid.to_owned())
}

fn parse_session_messages<R: std::io::Read>(reader: R) -> SessionHistory {
    let mut out = Vec::new();
    let mut compaction_count = 0_u32;
    let mut seen = std::collections::HashSet::new();
    // Queued prompts held until the row that says the CLI took one. The
    // attachment names the press; the remove row names the start, and that is
    // where the live path drew the words.
    let mut queued = Vec::new();
    for (idx, line_res) in BufReader::new(reader).lines().enumerate() {
        let line = match line_res {
            Ok(l) => l,
            Err(e) => {
                // The compaction count is also truncated here, and it is
                // rendered hidden at zero, so without naming it a torn
                // read is indistinguishable from a session that never
                // compacted. The transcript dir is file-sync replicated,
                // so a torn read is a live case.
                tracing::warn!(
                    line_no = idx,
                    error = %e,
                    compactions_counted_before_truncation = compaction_count,
                    "session scan: read failed; truncating both the message list and the compaction count"
                );
                break;
            }
        };
        if line.is_empty() {
            continue;
        }
        let value = match serde_json::from_str::<Value>(&line) {
            Ok(v) => v,
            Err(e) => {
                tracing::debug!(
                    line_no = idx,
                    error = %e,
                    "session scan: skipping unparseable line"
                );
                continue;
            }
        };
        if let Some(remove) = queue_remove_row(&value) {
            let row = queued_start_row(&mut queued, &value, remove);
            if !already_kept(&mut seen, &row.uuid) {
                out.push(row);
            }
            continue;
        }
        let Some((message, boundary)) = transcript_row(&value) else {
            continue;
        };
        if is_queued_attachment_row(&value) {
            queued.push(message);
            continue;
        }
        if already_kept(&mut seen, &message.uuid) {
            continue;
        }
        if boundary {
            compaction_count = compaction_count.saturating_add(1);
        }
        out.push(message);
    }
    // A prompt with no remove row never started: the live path drew no row
    // for it, and neither does the read.
    SessionHistory { messages: out, compaction_count }
}

/// Whether a transcript line is the attachment the CLI writes for a prompt
/// queued mid-turn.
///
/// **The harness's own completion notices are not these.** A background
/// task's notice is pressed into the turn that was already running - it was
/// never waiting in the queue - so its attachment row is its row, where the
/// run ended. Only a prompt the CLI took moves to its remove row, at the
/// turn's end or absorbed into the running turn.
fn is_queued_attachment_row(value: &Value) -> bool {
    if value.get("type").and_then(Value::as_str) != Some("attachment") {
        return false;
    }
    let Some(attachment) = value.get("attachment") else { return false };
    attachment.get("type").and_then(Value::as_str) == Some("queued_command")
        && attachment.get("commandMode").and_then(Value::as_str) != Some("task-notification")
}

/// The removal reasons that mean the CLI TOOK the prompt, read off this
/// machine's transcripts: `absorbed_mid_turn` (pressed into the running
/// turn), `delivered_to_agent` (handed over at its boundary), and the empty
/// reason on rows written before the field existed. Any other reason did not
/// start the prompt and draws no row.
const START_REASONS: [&str; 3] = ["absorbed_mid_turn", "delivered_to_agent", ""];

/// The `queue-operation remove` line as the queued prompt's row, or `None`
/// for any other line.
///
/// **The remove is the start.** The CLI writes it when it takes the prompt
/// out of its queue - at the end of the turn that was running, or absorbed
/// into it - so the row's clock is that removal, not the press. Only a
/// reason in [`START_REASONS`] takes the prompt.
fn queue_remove_row(value: &Value) -> Option<SessionMessage> {
    if value.get("type").and_then(Value::as_str) != Some("queue-operation")
        || value.get("operation").and_then(Value::as_str) != Some("remove")
    {
        return None;
    }
    let reason = value.get("reason").and_then(Value::as_str).unwrap_or_default();
    if !START_REASONS.contains(&reason) {
        return None;
    }
    let content = value.get("content").and_then(Value::as_str)?;
    // A background task's notice was absorbed into the running turn, so its
    // attachment already drew the row where the task ended; the removal is
    // not a prompt start and draws nothing.
    if content.trim_start().starts_with("<task-notification>") {
        return None;
    }
    let attachment = serde_json::json!({ "prompt": content });
    Some(SessionMessage {
        kind: SessionMessageKind::User,
        uuid: queue_row_id(value),
        session_id: value.get("sessionId").and_then(Value::as_str).unwrap_or_default().to_owned(),
        message: synthesize_queued_command_message(&attachment),
        parent_tool_use_id: None,
        timestamp: value.get("timestamp").and_then(Value::as_str).map(str::to_owned),
        tool_use_result: None,
        synthetic: false,
    })
}

/// The row's id, minted from the one line an anchored read can reproduce.
fn queue_row_id(value: &Value) -> String {
    use std::hash::{Hash as _, Hasher as _};
    let timestamp = value.get("timestamp").and_then(Value::as_str).unwrap_or_default();
    let content = value.get("content").and_then(Value::as_str).unwrap_or_default();
    let mut hasher = std::collections::hash_map::DefaultHasher::new();
    content.hash(&mut hasher);
    format!("queue-remove-{timestamp}-{:x}", hasher.finish())
}

/// The prompt words an attachment row's synthesised message carries.
fn queued_prompt_text(row: &SessionMessage) -> Option<&str> {
    row.message.get("content").and_then(Value::as_array)?.iter().find_map(|block| {
        (block.get("type")?.as_str()? == "queued_command")
            .then(|| block.get("prompt")?.as_str())
            .flatten()
    })
}

/// The row a remove line draws, taking the attachment that named the press if
/// one is waiting.
///
/// The attachment is what carries a multi-modal prompt's blocks, its
/// `commandMode` and the `source_uuid` the live row pairs with; the remove
/// line contributes the position, the clock and the id.
fn queued_start_row(
    queued: &mut Vec<SessionMessage>,
    remove: &Value,
    mut row: SessionMessage,
) -> SessionMessage {
    let content = remove.get("content").and_then(Value::as_str).unwrap_or_default();
    // **Exact words only.** A single pending attachment is not a safe guess:
    // two prompts can be in play, and pairing by position would draw one
    // prompt's words under another prompt's start.
    let at = queued.iter().position(|held| queued_prompt_text(held) == Some(content));
    if let Some(at) = at {
        let attachment = queued.remove(at);
        row.message = attachment.message;
        row.parent_tool_use_id = attachment.parent_tool_use_id;
        row.synthetic = attachment.synthetic;
        if row.session_id.is_empty() {
            row.session_id = attachment.session_id;
        }
    }
    row
}

/// Whether a transcript row carries this frame.
///
/// **The row rule's other half, and the one place it is stated.** A frame
/// whose kind no row round-trips - a `Result`, a thinking-token frame, a
/// `system` subtype whose row belongs to another surface - is one the live
/// stream emits and the CLI never writes, so a copy holding it counts a frame
/// the file counts no row for. That difference is what a page below the floor
/// has to be numbered through, by whoever still holds the frames the drops
/// took.
///
/// **A live hook frame has no row, though the read now serves hook rows.** The
/// CLI mints three ids for one run - the attachment row's uuid, the live
/// `hook_id` and the live frame's uuid - and none of them equals another, so a
/// live hook frame names no row a walk could resolve; the row the read
/// synthesizes is a copy of the run, not the frame. The answer stays false,
/// which is what keeps a stop off a frame the file cannot be asked for.
pub fn has_a_transcript_row(message: &forge_primitives::Message) -> bool {
    use forge_primitives::Message;
    // The row rule drops a sub-agent's frame, so a frame naming one has no
    // row either, whatever its kind: a numbering that counted it as having
    // one would be off for every page below it.
    let parent = match message {
        Message::User { parent_tool_use_id, .. }
        | Message::Assistant { parent_tool_use_id, .. }
        | Message::StopHookSummary { parent_tool_use_id, .. } => parent_tool_use_id.as_deref(),
        // The session task stamps a sub-agent's refusal with the dispatch it
        // belongs under, and the row rule drops it like any parented frame.
        Message::PermissionDenied { extras, .. } => {
            extras.get("parent_tool_use_id").and_then(Value::as_str)
        }
        _ => None,
    };
    if forge_primitives::names_a_dispatch(parent) {
        return false;
    }
    match message {
        Message::User { .. }
        | Message::Assistant { .. }
        | Message::StopHookSummary { .. }
        | Message::CompactBoundary { .. }
        | Message::PermissionDenied { .. }
        | Message::Notification { .. }
        // Typed system subtypes whose rows the read keeps: the conversation
        // draws a turn duration, and a dispatch's progress is read off the
        // row the scan keeps, so the numbering must count it.
        | Message::TurnDuration { .. }
        | Message::TaskProgress { .. } => true,
        // A generic `system` row has a row unless its subtype is one whose
        // row is drawn elsewhere: the read keeps the set the client draws.
        Message::System { subtype, .. } => {
            !SYSTEM_ROWS_DRAWN_ELSEWHERE.contains(&subtype.as_str())
        }
        // An unknown top-level frame has a row unless its type is one of the
        // transcript's own bookkeeping kinds, which no live stream emits.
        Message::Unknown { type_str, .. } => !UNSERVED_ROWS.contains(&type_str.as_str()),
        _ => false,
    }
}

/// The `system` subtypes whose row is not the conversation's: the task
/// lifecycle lands on its call's row, the background-task and command lists
/// feed their own surfaces, the handshake is not content, and a hook run
/// draws from its attachment row. **The client filters the same list** (its
/// own `SYSTEM_DRAWN_ELSEWHERE`), because the read and the live path are one
/// obligation - a subtype added here belongs there, and the other way round.
const SYSTEM_ROWS_DRAWN_ELSEWHERE: [&str; 11] = [
    "task_started",
    "task_updated",
    "task_notification",
    "background_tasks_changed",
    "commands_changed",
    "init",
    "status",
    "thinking_tokens",
    "hook_started",
    "hook_progress",
    "hook_response",
];

/// Top-level rows the read does not serve, of two kinds: the transcript's
/// own bookkeeping, which the CLI writes to disk and never streams, and the
/// wire types that live only on the stream and have no durable row.
///
/// **A row not named here is an unknown type, and draws as itself.** These
/// are the types the read has always dropped, so keeping one would change
/// what a reopened conversation holds rather than restore what a live one
/// drew.
const UNSERVED_ROWS: [&str; 18] = [
    // Bookkeeping the CLI writes and never streams.
    "last-prompt",
    "atis-latch",
    "queue-operation",
    "worktree-state",
    "mode",
    "pr-link",
    "tag",
    "permission-mode",
    "cost-state",
    "ai-title",
    "file-history-snapshot",
    "summary",
    // Wire-only kinds: a row of one of these is a fixture's or an artefact's,
    // not something the CLI persists.
    "result",
    "error",
    "rate_limit_event",
    "command_lifecycle",
    "stream_event",
    "tool_progress",
];

/// One transcript row as the session's message, or `None` for a row that is
/// not one. The flag says the row is a compaction boundary, which the read
/// counts.
///
/// **The one row rule, shared by every read of a transcript.** A resume walks
/// the whole file and the paging span read walks a window of it, and a second
/// copy of this is how the two would come to type the same row differently.
fn transcript_row(value: &Value) -> Option<(SessionMessage, bool)> {
    let row_type = value.get("type").and_then(Value::as_str);
    let mut boundary = false;
    let (kind, message) = match row_type {
        Some("user") => (SessionMessageKind::User, value.get("message").cloned()),
        Some("assistant") => (SessionMessageKind::Assistant, value.get("message").cloned()),
        // Attachment rows hold claude's persisted record of mid-turn
        // queued inputs - `{"type":"attachment", "attachment":{"type":
        // "queued_command", "prompt":"...", "commandMode":"prompt"}}`.
        // On replay we hoist them into a synthetic user envelope
        // whose single content block is the `queued_command`, so the
        // downstream walker reconstructs the user bubble that was
        // never on the wire as a regular user message.
        Some("attachment") => match value.get("attachment") {
            Some(att) if att.get("type").and_then(Value::as_str) == Some("queued_command") => {
                (SessionMessageKind::User, Some(synthesize_queued_command_message(att)))
            }
            // **A hook's own record is the row its live frames drew.** The
            // live stream sends a run as `system/hook_*` frames and the
            // transcript persists it as an attachment; dropping the row left a
            // page without rows the live conversation drew (rule 25), and
            // every later landing reconciled against a copy that had none of
            // them.
            //
            // **The served set is the live wire's**: the SessionStart and
            // Setup runs it gates live hook frames on, in their success,
            // cancelled and non-blocking-error records. Every other event's
            // rows stay the CLI's own context, and `hook_additional_context`
            // is not a run at all - the live path drew none of them as a row,
            // so a read that did would show a reader rows they never had.
            Some(att) if is_hook_attachment(att) => {
                (SessionMessageKind::System, Some(synthesize_hook_frame(value, att)))
            }
            _ => return None,
        },
        // The only durable record of how often this session has compacted
        // and of where each cut fell - nothing else survives a resume.
        // Counted by the caller, and the row is kept so a resumed conversation
        // draws its boundaries the way a live one does.
        Some("system")
            if value.get("subtype").and_then(Value::as_str) == Some("compact_boundary") =>
        {
            boundary = true;
            // Keyed on what the row yielded rather than on the metadata
            // object being there: the plausible drift is a rename inside it
            // (`preTokens`, per the primitives test), which leaves the
            // object present and one field unread - a row drawn without
            // that fact, and nothing else saying why. The live arm warns on
            // the same degradation. Every modelled field is read here, so a
            // rename of any one of them lands in this record.
            let missing = missing_boundary_facts(value);
            if !missing.is_empty() {
                let uuid = value.get("uuid").and_then(Value::as_str).unwrap_or_default();
                let session = session_in_row(value);
                tracing::warn!(
                    target: "forge_agent::userdata::catalog",
                    event_name = "compact_boundary_missing_facts",
                    uuid,
                    session,
                    missing = %missing.join(", "),
                    "compact_boundary row did not yield every fact; counted, but the kept frame arrives untyped",
                );
            }
            (SessionMessageKind::System, Some(compact_boundary_frame(value)))
        }
        // What the turn's hooks did. A system row is the frame the wire
        // sends, so it is kept whole; the one field the frame's decoder
        // spells differently is the session.
        Some("system")
            if value.get("subtype").and_then(Value::as_str) == Some("stop_hook_summary") =>
        {
            (SessionMessageKind::System, Some(wire_session_frame(value)))
        }
        // **The system rows the live conversation draws.** A refusal, a
        // failed API call, an informational line, a notification and any
        // other subtype whose row is the conversation's are each a line the
        // session was told and the CLI persists, and the scan dropped every
        // system row but the two above - so a resumed seat or a reopened page
        // lost frames the live view had (rule 25). Kept whole, in the shape
        // the wire sends, with the wire's own spelling of the session stamped
        // on. A row whose subtype draws elsewhere keeps dropping here, the
        // same list the client filters.
        Some("system")
            if value
                .get("subtype")
                .and_then(Value::as_str)
                .is_none_or(|subtype| !SYSTEM_ROWS_DRAWN_ELSEWHERE.contains(&subtype)) =>
        {
            (SessionMessageKind::System, Some(wire_session_frame(value)))
        }
        // **A top-level row this build has no kind for draws as itself.** The
        // row is kept whole and decoded by the same decoder the wire feeds -
        // `messages_from_rows` presents it under the `system` arm, the one
        // that hands a frame straight to `Message`. The rows the read has
        // never served stay out (see [`UNSERVED_ROWS`]).
        Some(other) if other != "system" && !UNSERVED_ROWS.contains(&other) => {
            (SessionMessageKind::System, Some(value.clone()))
        }
        _ => return None,
    };
    // The same rule the fold reads a frame by: a parent id that names a
    // dispatch, which on the wire is never empty and never null.
    if value
        .get("parent_tool_use_id")
        .and_then(Value::as_str)
        .is_some_and(|parent| !parent.trim().is_empty())
    {
        return None;
    }
    let uuid = value.get("uuid").and_then(Value::as_str).unwrap_or_default().to_string();
    let sess = session_in_row(value);
    let timestamp = value.get("timestamp").and_then(Value::as_str).map(str::to_owned);
    // The CLI's own stamps that nobody typed this row. They do not
    // co-occur - the reminder carries `isMeta` and `turnCompanion`, a
    // compaction summary carries `isCompactSummary` and
    // `isVisibleInTranscriptOnly` and no `isMeta` - and the wire writes
    // the whole set as one `isSynthetic`, so reading a subset here is
    // what let a summary be the reader's turn on resume and the harness's
    // line live.
    let synthetic = ["isMeta", "isCompactSummary", "isVisibleInTranscriptOnly", "turnCompanion"]
        .iter()
        .any(|flag| value.get(flag).and_then(Value::as_bool).unwrap_or(false));
    Some((
        SessionMessage {
            kind,
            uuid,
            session_id: sess,
            message: message.unwrap_or(Value::Null),
            parent_tool_use_id: None,
            timestamp,
            tool_use_result: value.get("toolUseResult").filter(|result| !result.is_null()).cloned(),
            synthetic,
        },
        boundary,
    ))
}

/// The session a transcript row belongs to. The wire spells it
/// `session_id`; a row the CLI wrote spells it `sessionId`.
fn session_in_row(value: &Value) -> String {
    value
        .get("session_id")
        .or_else(|| value.get("sessionId"))
        .and_then(Value::as_str)
        .unwrap_or_default()
        .to_owned()
}

/// A kept row carrying the session under the spelling the live frame sends.
///
/// **The CLI writes `sessionId` and the wire sends `session_id`.** Stamping
/// the wire's spelling onto a clone that still held the CLI's put both in the
/// frame, so the raw a reader opens showed the session twice (Ved, 2026-10-28).
fn wire_session_frame(value: &Value) -> Value {
    let mut frame = value.clone();
    if let Some(record) = frame.as_object_mut() {
        record.remove("sessionId");
        record.insert("session_id".into(), Value::String(session_in_row(value)));
    }
    frame
}

/// Which of the facts the kept boundary frame reads the row did not yield -
/// the trigger, and the counts either side of the cut.
///
/// Read off the same three fields `compact_boundary_frame` builds from, and
/// named rather than counted because the record built from this is the only
/// thing that would say which disk key moved.
fn missing_boundary_facts(value: &Value) -> Vec<&'static str> {
    let metadata = value.get("compactMetadata");
    let read = |key: &str| metadata.and_then(|metadata| metadata.get(key));
    [
        ("trigger", read("trigger").and_then(Value::as_str).is_none()),
        ("preTokens", read("preTokens").and_then(Value::as_u64).is_none()),
        ("postTokens", read("postTokens").and_then(Value::as_u64).is_none()),
    ]
    .into_iter()
    .filter_map(|(fact, absent)| absent.then_some(fact))
    .collect()
}

/// The boundary row in the shape the wire sends.
///
/// A transcript spells its metadata flat and camelCase (`compactMetadata`,
/// `preTokens`), while the decoder keys on the wire's nesting - so a row handed
/// on as read decodes as a generic system frame, and draws bare or half-filled
/// where a live boundary draws its counts. The row's own stamp rides with it:
/// the frame is built here rather than copied, so this is where a page's clock
/// would lose the one fact only the file has.
fn compact_boundary_frame(value: &Value) -> Value {
    let metadata = value.get("compactMetadata");
    let field =
        |key: &str| metadata.and_then(|metadata| metadata.get(key)).cloned().unwrap_or(Value::Null);
    let mut frame = serde_json::json!({
        "type": "system",
        "subtype": "compact_boundary",
        "session_id": session_in_row(value),
        "uuid": value.get("uuid").and_then(Value::as_str).unwrap_or_default(),
        "compact_metadata": {
            "trigger": field("trigger"),
            "pre_tokens": field("preTokens"),
            "post_tokens": field("postTokens"),
        },
    });
    if let Some(timestamp) = value.get("timestamp") {
        frame["timestamp"] = timestamp.clone();
    }
    frame
}

/// Build a `{"role":"user","content":[{queued_command}]}` envelope from
/// a JSONL attachment row's `attachment` field. Used during session
/// replay to surface mid-turn queued inputs that claude persisted as
/// `type:"attachment"` rows (which the scanner would otherwise skip).
fn synthesize_queued_command_message(attachment: &Value) -> Value {
    let mut block = serde_json::Map::new();
    block.insert("type".into(), Value::String("queued_command".into()));
    if let Some(prompt) = attachment.get("prompt") {
        block.insert("prompt".into(), prompt.clone());
    }
    if let Some(mode) = attachment.get("commandMode") {
        block.insert("commandMode".into(), mode.clone());
    }
    if let Some(src) = attachment.get("source_uuid") {
        block.insert("source_uuid".into(), src.clone());
    }
    serde_json::json!({
        "role": "user",
        "content": [Value::Object(block)],
    })
}

/// The hook events whose runs the live stream forwards, and therefore the only
/// ones a read serves as rows.
///
/// **The CLI gates live hook frames on these outside remote.** A page that
/// drew any other event would show rows the live view never had - a divergence
/// the read would introduce rather than repair - so the rest of the file's
/// hook rows stay context the CLI keeps for itself.
fn live_hook_event(attachment: &Value) -> bool {
    matches!(attachment.get("hookEvent").and_then(Value::as_str), Some("SessionStart" | "Setup"))
}

/// Whether an attachment row records a hook run the live stream forwards.
///
/// The persisted run records are the success, the cancelled run and the
/// non-blocking error; `hook_additional_context` is aggregate context rather
/// than a run, and the live path drew it as none.
fn is_hook_attachment(attachment: &Value) -> bool {
    live_hook_event(attachment)
        && matches!(
            attachment.get("type").and_then(Value::as_str),
            Some("hook_success" | "hook_cancelled" | "hook_non_blocking_error")
        )
}

/// One persisted hook run as the live frame the chat folds.
///
/// **The two carriers' ids never meet.** The CLI stamps the row a uuid of its
/// own, the live run a `hook_id` and the live frame another uuid - three
/// distinct ids for one run - so a read copy that borrowed any of them would
/// still name nothing the held frame names. The row keeps its own uuid, and
/// `transcript` marks it as the read's copy: the client pairs it with a held
/// live run of the same event and name, and keeps one row.
fn synthesize_hook_frame(row: &Value, attachment: &Value) -> Value {
    let mut frame = serde_json::Map::new();
    frame.insert("type".into(), Value::String("system".into()));
    frame.insert("subtype".into(), Value::String("hook_response".into()));
    if let Some(id) = row.get("uuid") {
        frame.insert("hook_id".into(), id.clone());
        frame.insert("uuid".into(), id.clone());
    }
    for (field, from) in [("hook_name", "hookName"), ("hook_event", "hookEvent")] {
        if let Some(held) = attachment.get(from) {
            frame.insert(field.into(), held.clone());
        }
    }
    // **The live frame's own fields, all of them.** `HookResponse` in the
    // primitives requires every one, and a cancelled row carries none of the
    // stream fields at all: a frame that omitted them would decode as the
    // generic system shape rather than the typed response the live wire sends.
    // Absent ones are the empty value the live frame's own type carries.
    frame.insert("output".into(), Value::String(hook_output(attachment).unwrap_or_default()));
    for field in ["stdout", "stderr"] {
        let held = attachment.get(field).and_then(Value::as_str).unwrap_or_default();
        frame.insert(field.into(), Value::String(held.to_owned()));
    }
    let code = attachment.get("exitCode").and_then(Value::as_i64).unwrap_or(0);
    frame.insert("exit_code".into(), Value::Number(code.into()));
    frame.insert("outcome".into(), hook_outcome(attachment));
    frame.insert("session_id".into(), Value::String(session_in_row(row)));
    frame.insert("transcript".into(), Value::Bool(true));
    Value::Object(frame)
}

/// The outcome the live frame reports for one persisted run record.
///
/// A timeout is persisted as a cancellation with `timedOut`, and draws as a
/// cancelled run; a non-blocking error is the error outcome the live path
/// sends for it.
fn hook_outcome(attachment: &Value) -> Value {
    let outcome = match attachment.get("type").and_then(Value::as_str) {
        Some("hook_non_blocking_error") => "error",
        Some("hook_cancelled") => "cancelled",
        _ => "success",
    };
    Value::String(outcome.to_owned())
}

/// What one persisted hook run printed, as a `hook_response` frame carries it.
///
/// `content` is the combined output, a string or the lines of one; the raw
/// streams are the fallback, `stdout` first, so a row whose combined field is
/// empty still draws what the hook wrote. Both branches read the same way: a
/// value that is only whitespace is nothing to draw.
fn hook_output(attachment: &Value) -> Option<String> {
    let held = match attachment.get("content") {
        Some(Value::String(text)) => Some(text.clone()),
        Some(Value::Array(lines)) => {
            Some(lines.iter().filter_map(Value::as_str).collect::<Vec<_>>().join("\n"))
        }
        _ => None,
    };
    held.into_iter()
        .chain(
            ["stdout", "stderr"].iter().filter_map(|field| {
                attachment.get(field).and_then(Value::as_str).map(str::to_owned)
            }),
        )
        .find(|text| !text.trim().is_empty())
}

/// Sanitise a path the same way the `claude` CLI does -
/// non-alphanumerics become hyphens, and overlong paths are
/// truncated with a base-36 hash suffix (matching JS's
/// `String.prototype.hashCode` trick).
fn sanitize_path(name: &str) -> String {
    let sanitized: String =
        name.chars().map(|c| if c.is_ascii_alphanumeric() { c } else { '-' }).collect();
    if sanitized.len() <= MAX_SANITIZED_LENGTH {
        return sanitized;
    }
    let hash = simple_hash(name);
    let truncated: String = sanitized.chars().take(MAX_SANITIZED_LENGTH).collect();
    format!("{truncated}-{hash}")
}

/// 32-bit integer hash to base-36, matching the CLI's directory naming.
fn simple_hash(s: &str) -> String {
    let mut h: i64 = 0;
    for ch in s.chars() {
        let c = ch as i64;
        h = (h << 5).wrapping_sub(h).wrapping_add(c);
        // Emulate JS `hash |= 0` (coerce to 32-bit signed int)
        h &= 0xFFFF_FFFF;
        if h >= 0x8000_0000 {
            h -= 0x1_0000_0000;
        }
    }
    let mut n = h.unsigned_abs();
    if n == 0 {
        return "0".into();
    }
    let digits = b"0123456789abcdefghijklmnopqrstuvwxyz";
    let mut out = Vec::new();
    while n > 0 {
        out.push(digits[(n % 36) as usize]);
        n /= 36;
    }
    out.reverse();
    String::from_utf8(out).unwrap_or_default()
}

fn project_dir_for(config_dir: &Path, project_path: &str) -> PathBuf {
    projects_dir_for(config_dir).join(sanitize_path(&canonicalize_path(project_path)))
}

/// Bounded-concurrency cap for [`list_sessions`] per-file lite reads.
///
/// Each transcript costs an open, a head read, a tail seek, and a tail
/// read, plus UTF-8 validation. On a project-rich install (~50 projects,
/// ~10 sessions each) the serial walk dominates connect time; capping
/// at 16 keeps fd pressure low while still saturating an SSD.
const LIST_SESSIONS_MAX_CONCURRENT: usize = 16;

/// True when `info` represents a worker session: one whose transcript
/// carries a `forge:worker:<label>` tag.
pub fn should_exclude_worker_tag(info: &SDKSessionInfo) -> bool {
    info.tag.as_deref().is_some_and(|t| t.starts_with(FORGE_WORKER_TAG_PREFIX))
}

/// Which worker sessions a listing may carry.
///
/// The seat decides, not the directory: a non-git worker runs in the
/// project root, so its own sessions sit in the same directory as its
/// lead's.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Workers {
    /// None. A lead's listing is the project's own sessions and never
    /// its workers', and the boot catalog wants the same.
    Hidden,
    /// Only the transcripts tagged for `label`: what a worker ran under
    /// its own label is what it can resume.
    Only(String),
}

/// List sessions. When `directory` is `Some`, scans that project dir;
/// when `None`, scans every project directory under `config_dir`'s
/// `projects/` tree. A capped listing stats each candidate for its mtime,
/// reads newest first on the tokio blocking pool (at most 16 at once), and
/// stops once the rows the cap serves are in hand - a short batch is filled
/// by the next, so the cap bounds the reads without shortening the listing.
/// An uncapped listing (the boot scan) reads every candidate. `workers`
/// decides which worker-tagged rows may appear, and it lands before the
/// cap: a listing is never shortened by rows it would not have carried.
///
/// # Panics
///
/// Never - filesystem errors fall through and produce an empty Vec.
///
/// `tag_cache` carries the previous run's tag scans so an unchanged
/// transcript is not re-read end to end. `None` reads every byte of
/// every file, which is what a caller with no store must do. A worker's
/// listing also skips the READ of a transcript that cache tags for another
/// seat: its tag cannot match, and the head-and-tail read is the cost a
/// non-git worker on a large project paid on every seat open (#2567).
pub async fn list_sessions(
    config_dir: &Path,
    directory: Option<&str>,
    limit: Option<usize>,
    offset: usize,
    workers: Workers,
    tag_cache: Option<&std::sync::Arc<SessionTagCache>>,
) -> Vec<SDKSessionInfo> {
    let search_dirs: Vec<PathBuf> = if let Some(dir) = directory {
        vec![project_dir_for(config_dir, dir)]
    } else {
        try_read_dir(&projects_dir_for(config_dir))
            .map(|iter| {
                iter.flatten()
                    .filter(|e| e.file_type().is_ok_and(|t| t.is_dir()))
                    .map(|e| e.path())
                    .collect()
            })
            .unwrap_or_default()
    };

    // Cheap directory walks first - collect every candidate path
    // synchronously. The expensive part is the per-file lite read,
    // which we hand off to spawn_blocking below.
    let mut candidates: Vec<PathBuf> = Vec::new();
    for project_dir in search_dirs {
        let Some(iter) = try_read_dir(&project_dir) else {
            continue;
        };
        for entry in iter.flatten() {
            let path = entry.path();
            if path.extension().and_then(|e| e.to_str()) == Some("jsonl") {
                candidates.push(path);
            }
        }
    }

    // **A worker's listing does not read what its own tag already rules out**
    // (#2567). The head-and-tail read of every transcript in the directory is
    // the whole cost of a `/resume` list on a non-git worker, which shares
    // the project root with its lead; an unchanged file the cache tags for
    // another seat (or for none) is dropped here instead. The answer is the
    // same rows either way, so the skip names itself on the log.
    if let (Workers::Only(label), Some(cache)) = (&workers, tag_cache) {
        let wanted = worker_tag(label);
        let before = candidates.len();
        // `None` is a tag this cache cannot vouch for, and that file is read.
        candidates.retain(|path| cache.carries_tag(path, &wanted).unwrap_or(true));
        if before > candidates.len() {
            // **No explicit target: the shipped directive names this module's
            // own path**, and a record under another target is filtered out
            // before it ever reaches `forge.log` (#2567).
            tracing::debug!(
                event_name = "session_scan_narrowed",
                dir = directory.unwrap_or("<every project>"),
                label = %label,
                skipped = before - candidates.len(),
                to_read = candidates.len(),
                "transcripts whose cached tag rules them out were not read",
            );
        }
    }

    // **A capped listing reads newest first and stops when it is full**
    // (#2567). Reading every candidate and cutting afterwards made a
    // long-lived worker label pay a head-and-tail read per file however few
    // rows it served; the stat here is the mtime the final order already
    // reads, so nothing is paid for an order the answer did not need.
    let mut ordered: Vec<(PathBuf, u64)> = candidates
        .into_iter()
        .map(|path| {
            let mtime = std::fs::metadata(&path)
                .and_then(|meta| meta.modified())
                .ok()
                .and_then(|time| time.duration_since(std::time::UNIX_EPOCH).ok())
                .map_or(0, |age| u64::try_from(age.as_millis()).unwrap_or(u64::MAX));
            (path, mtime)
        })
        .collect();
    ordered.sort_by_key(|(path, mtime)| (std::cmp::Reverse(*mtime), path.clone()));

    let wanted_tag = match &workers {
        Workers::Only(label) => Some(worker_tag(label)),
        Workers::Hidden => None,
    };
    let matches = |info: &SDKSessionInfo| match &wanted_tag {
        Some(wanted) => info.tag.as_deref() == Some(wanted.as_str()),
        None => !should_exclude_worker_tag(info),
    };
    // `None` is no cap: the boot scan reads the whole tree, which is what
    // its catalog is.
    let wanted_rows = limit.map(|l| offset.saturating_add(l));

    let mut entries: Vec<SDKSessionInfo> = Vec::new();
    let mut paths = ordered.into_iter();
    let mut set: tokio::task::JoinSet<Option<SDKSessionInfo>> = tokio::task::JoinSet::new();
    let spawn_read = |set: &mut tokio::task::JoinSet<Option<SDKSessionInfo>>, path: PathBuf| {
        let cache = tag_cache.cloned();
        set.spawn_blocking(move || read_session_info(&path, cache.as_deref()));
    };
    // **The reads stay pipelined, and the cap is counted rather than
    // re-counted.** The boot scan is uncapped and reads the whole tree, so
    // recounting the kept rows on every pass would multiply its cost by the
    // number of batches, and draining the set before refilling would leave
    // the pool idle between them.
    let mut kept = 0usize;
    loop {
        let owed = wanted_rows.map_or(usize::MAX, |wanted| wanted.saturating_sub(kept));
        if owed == 0 {
            break;
        }
        // **A batch is no wider than the rows still owed**, so a cap reads
        // the cap and not a pool's width past it. A short batch - a
        // transcript gone, or one the seat's own tag rules out - is filled by
        // the next, so the read is bounded without the listing coming up
        // short.
        let width = owed.min(LIST_SESSIONS_MAX_CONCURRENT);
        while set.len() < width {
            let Some((path, _)) = paths.next() else { break };
            spawn_read(&mut set, path);
        }
        let Some(res) = set.join_next().await else { break };
        match res {
            Ok(Some(info)) => {
                if matches(&info) {
                    kept += 1;
                }
                entries.push(info);
            }
            Ok(None) => {}
            Err(err) => {
                tracing::debug!(
                    target: "forge_agent::catalog::scan",
                    error = %err,
                    "list_sessions: per-file read task failed",
                );
            }
        }
    }
    if let Some(wanted) = wanted_rows {
        tracing::debug!(
            event_name = "session_scan_read",
            dir = directory.unwrap_or("<every project>"),
            read = entries.len(),
            wanted,
            "the listing read the rows its cap serves, newest first",
        );
    }

    entries.sort_by_key(|e| std::cmp::Reverse(e.last_modified));
    let entries: Vec<SDKSessionInfo> = entries.into_iter().filter(|info| matches(info)).collect();
    let end = limit.map_or(entries.len(), |l| offset.saturating_add(l));
    entries.into_iter().skip(offset).take(end.saturating_sub(offset)).collect()
}

/// Tail shared by every `get_session_messages` bail. The compaction count
/// these paths return is unknown rather than zero, and the Projects-pane
/// row hides at zero, so "count says 0 on a session I know compacted"
/// has no triage path unless each bail is greppable.
const UNREAD_TRANSCRIPT_COUNT_NOTE: &str =
    "transcript unread, so the compaction count is unknown rather than zero";

/// Read the full transcript for one session.
///
/// Returns a default [`SessionHistory`] when the file can't be found or
/// opened - no messages and a **zero** compaction count, which is
/// indistinguishable from a session that never compacted. Each of those
/// paths logs `UNREAD_TRANSCRIPT_COUNT_NOTE` so the difference is
/// recoverable from the log even though it is not from the return value.
pub fn get_session_messages(
    config_dir: &Path,
    session_id: &str,
    directory: Option<&str>,
) -> SessionHistory {
    read_session_messages(config_dir, session_id, directory, true)
}

/// [`get_session_messages`] over the session's own rows only: no sub-agent
/// sidecars are opened or folded. The handoff brief takes this one, so it
/// neither pays for the sidecar parse nor carries frames it does not read. A
/// page read does not: a dispatch row draws its sidecar frames (#1763).
pub(crate) fn read_session_messages_main_thread(
    config_dir: &Path,
    session_id: &str,
    directory: Option<&str>,
) -> SessionHistory {
    read_session_messages(config_dir, session_id, directory, false)
}

fn read_session_messages(
    config_dir: &Path,
    session_id: &str,
    directory: Option<&str>,
    fold_sidecars: bool,
) -> SessionHistory {
    if !is_valid_uuid(session_id) {
        tracing::warn!(
            %session_id,
            "get_session_messages: not a session uuid; {UNREAD_TRANSCRIPT_COUNT_NOTE}",
        );
        return SessionHistory::default();
    }
    let file_name = format!("{session_id}.jsonl");
    let candidate = if let Some(dir) = directory {
        Some(project_dir_for(config_dir, dir).join(&file_name))
    } else {
        try_read_dir(&projects_dir_for(config_dir)).and_then(|iter| {
            iter.flatten().map(|e| e.path().join(&file_name)).find(|p| p.is_file())
        })
    };
    let Some(path) = candidate else {
        tracing::debug!(
            %session_id,
            "get_session_messages: no on-disk file for session; {UNREAD_TRANSCRIPT_COUNT_NOTE}",
        );
        return SessionHistory::default();
    };
    let file = match fs::File::open(&path) {
        Ok(file) => file,
        Err(err) => {
            tracing::warn!(
                %session_id,
                path = %path.display(),
                %err,
                "get_session_messages: open failed; backfill renders empty and {UNREAD_TRANSCRIPT_COUNT_NOTE}",
            );
            return SessionHistory::default();
        }
    };
    let mut history = parse_session_messages(file);
    if fold_sidecars {
        let (sidecars, links) = subagent_sidecars(session_id, &path);
        fold_subagent_sidecars(&mut history.messages, sidecars, &links, session_id);
    }
    history
}

/// Insert each dispatch's sidecar frames under the first row its chain
/// answers. Appending them at the end would put them at the newest edge of
/// the resumed window, ahead of the main rows.
///
/// A nested dispatch is named by a call inside its parent's sidecar, never by
/// the transcript, so its chain is walked to the first dispatch the transcript
/// answers and every row is stamped with that card. Resolution happens before
/// any splice, which is what makes the child independent of the parent's read
/// order; a block whose parent body was read then emits inside the row that
/// fired it, and one whose parent was not read follows the owner's anchor.
fn fold_subagent_sidecars(
    messages: &mut Vec<SessionMessage>,
    sidecars: Vec<(String, Vec<SessionMessage>)>,
    links: &HashMap<String, SidecarLink>,
    session_id: &str,
) {
    // Every call an agent's sidecar holds was made under that sidecar's own
    // dispatch: the CLI writes no parent into a sidecar row. The metas carry
    // the same link without the parent's body, which is what keeps a nested
    // sidecar resolvable when its parent fell past the read budget.
    let mut made_under: HashMap<String, String> = HashMap::new();
    for (dispatch, rows) in &sidecars {
        for row in rows {
            for id in tool_use_ids(row) {
                made_under.insert(id.to_owned(), dispatch.clone());
            }
        }
    }
    for link in links.values() {
        let Some(parent) = link.parent_agent.as_deref() else {
            continue;
        };
        let Some(parent_link) = links.get(parent) else {
            continue;
        };
        made_under.insert(link.dispatch.clone(), parent_link.dispatch.clone());
    }
    // Only a transcript row anchors a block; a sidecar folded into another
    // sidecar's rows would nest an expansion inside an expansion. Tool-use
    // ids are unique per call, so a spliced row cannot answer another
    // sidecar's dispatch.
    let answered: HashSet<String> =
        messages.iter().flat_map(answered_dispatches).map(str::to_owned).collect();
    // Group by owner, then splice each owner's tree in the order it ran: a
    // nested block emits at the row carrying its dispatch, so a depth band
    // cannot flatten it. Nesting rides the tool_use ids; only roots follow
    // arrival order.
    let mut folded: HashMap<String, Vec<Block>> = HashMap::new();
    for (dispatch, mut rows) in sidecars {
        let Some(owner) = owning_dispatch(&dispatch, &answered, &made_under) else {
            // No row of the transcript answers the dispatch or any dispatch
            // it was nested under: there is no expansion for these frames.
            tracing::debug!(
                event_name = "subagent_sidecar_unresolved",
                session = %session_id,
                dispatch = %dispatch,
                "subagent sidecar skipped: no dispatch in its chain is answered by the transcript",
            );
            continue;
        };
        for row in &mut rows {
            row.parent_tool_use_id = Some(owner.to_owned());
        }
        folded.entry(owner.to_owned()).or_default().push(Block { dispatch, rows });
    }
    for (owner, mut blocks) in folded {
        // The sidecars arrive newest-first; roots follow arrival, so reverse
        // them. Nesting itself rides the tool_use ids, not the mtimes.
        blocks.reverse();
        // `owning_dispatch` only answers for a dispatch `answered` holds, and
        // `messages` only grows below, so this lookup cannot miss; the guard
        // keeps the splice total.
        let Some(index) = messages.iter().position(|row| answers_dispatch(row, &owner)) else {
            continue;
        };
        let rows = emit_sidecar_tree(blocks);
        messages.splice((index + 1)..=index, rows);
    }
}

/// One sidecar's frames and the dispatch it ran as.
struct Block {
    dispatch: String,
    rows: Vec<SessionMessage>,
}

/// One owner's blocks, flattened in the order the work ran: a block emits at
/// the row naming its dispatch, and a block no other names - its parent body
/// was past the budget - follows the roots.
fn emit_sidecar_tree(mut blocks: Vec<Block>) -> Vec<SessionMessage> {
    let by_dispatch: HashMap<String, usize> =
        blocks.iter().enumerate().map(|(at, block)| (block.dispatch.clone(), at)).collect();
    // A block whose dispatch names another block's row is a child; the rest
    // are roots.
    let mut inline = vec![false; blocks.len()];
    for (at, block) in blocks.iter().enumerate() {
        for row in &block.rows {
            for id in tool_use_ids(row) {
                if let Some(&kid) = by_dispatch.get(id)
                    && kid != at
                {
                    inline[kid] = true;
                }
            }
        }
    }
    let mut out = Vec::new();
    let mut emitted = vec![false; blocks.len()];
    for at in (0..blocks.len()).filter(|at| !inline[*at]) {
        emit_block(&mut blocks, at, &by_dispatch, &mut emitted, &mut out);
    }
    // A malformed link pair can make a block its own ancestor; it still
    // draws, once, after the roots rather than vanishing.
    for at in 0..blocks.len() {
        emit_block(&mut blocks, at, &by_dispatch, &mut emitted, &mut out);
    }
    out
}

/// Append one block, with each child block inserted after the row that
/// carries its dispatch.
fn emit_block(
    blocks: &mut [Block],
    at: usize,
    by_dispatch: &HashMap<String, usize>,
    emitted: &mut [bool],
    out: &mut Vec<SessionMessage>,
) {
    if emitted[at] {
        return;
    }
    emitted[at] = true;
    for row in std::mem::take(&mut blocks[at].rows) {
        let mut kids = Vec::new();
        for id in tool_use_ids(&row) {
            if let Some(&kid) = by_dispatch.get(id)
                && kid != at
            {
                kids.push(kid);
            }
        }
        out.push(row);
        for kid in kids {
            emit_block(blocks, kid, by_dispatch, emitted, out);
        }
    }
}

/// The dispatch whose row a sidecar's frames belong under: the first dispatch
/// in its chain the transcript answers.
fn owning_dispatch<'a>(
    dispatch: &'a str,
    answered: &HashSet<String>,
    made_under: &'a HashMap<String, String>,
) -> Option<&'a str> {
    let mut cursor = dispatch;
    // Bounded, so a pair of sidecars naming each other cannot spin: these
    // links are read off disk rather than built here.
    for _ in 0..=made_under.len() {
        if answered.contains(cursor) {
            return Some(cursor);
        }
        cursor = made_under.get(cursor)?;
    }
    None
}

/// The calls made in `row`.
fn tool_use_ids(row: &SessionMessage) -> impl Iterator<Item = &str> {
    block_field(row, "tool_use", "id")
}

/// The calls `row` answers.
fn answered_dispatches(row: &SessionMessage) -> impl Iterator<Item = &str> {
    block_field(row, "tool_result", "tool_use_id")
}

/// The `field` of every content block of `kind` in `row`.
fn block_field<'a>(
    row: &'a SessionMessage,
    kind: &'static str,
    field: &'static str,
) -> impl Iterator<Item = &'a str> {
    row.message.get("content").and_then(Value::as_array).into_iter().flatten().filter_map(
        move |block| {
            (block.get("type").and_then(Value::as_str) == Some(kind))
                .then(|| block.get(field).and_then(Value::as_str))
                .flatten()
        },
    )
}

/// Whether `row` is the tool result answering `dispatch`.
fn answers_dispatch(row: &SessionMessage, dispatch: &str) -> bool {
    answered_dispatches(row).any(|answered| answered == dispatch)
}

/// How many bytes of sidecar frames one resume reads in total. A session can
/// hold hundreds of sidecars and most of them are older than the window the
/// conversation is cut to, so the read is bounded rather than whole
/// (#1763).
const SIDECAR_BUDGET_BYTES: u64 = 8 * 1024 * 1024;

/// A sidecar's own meta: the dispatch its agent ran as, and the agent whose
/// sidecar fired that dispatch.
struct SidecarLink {
    dispatch: String,
    parent_agent: Option<String>,
}

/// The sidecar bodies the budget admitted, and every meta's chain link.
type SidecarRead = (Vec<(String, Vec<SessionMessage>)>, HashMap<String, SidecarLink>);

/// The frames a session's sub-agent sidecars hold, with each sidecar's own
/// dispatch and every meta's chain link. The links are read for all sidecars,
/// not only the ones inside the budget; the fold stamps the rows themselves.
///
/// The CLI writes a sub-agent's own frames to
/// `<session>/subagents/agent-<id>.jsonl` and names the dispatch and its
/// parent agent in the sibling `.meta.json`; neither is in the session
/// transcript, so a resumed conversation drew every card as its dispatch
/// alone until this read folded them in (#1763). Newest sidecars first, up
/// to [`SIDECAR_BUDGET_BYTES`]; a sidecar whose body cannot be read, or
/// falls past the budget, keeps the dispatch alone, while its meta still
/// links a nested child whose body is inside the budget. A meta that cannot
/// be parsed records no link.
fn subagent_sidecars(session_id: &str, transcript: &Path) -> SidecarRead {
    subagent_sidecars_with_budget(session_id, transcript, SIDECAR_BUDGET_BYTES)
}

/// [`subagent_sidecars`] with the read budget given, so the cap is drivable
/// at a size a test can build.
fn subagent_sidecars_with_budget(session_id: &str, transcript: &Path, budget: u64) -> SidecarRead {
    let Some(dir) = transcript.parent() else {
        return (Vec::new(), HashMap::new());
    };
    let sidecar_dir = dir.join(session_id).join("subagents");
    let entries = match fs::read_dir(&sidecar_dir) {
        Ok(entries) => entries,
        // No directory is the ordinary case: the session never dispatched.
        Err(error) if error.kind() == std::io::ErrorKind::NotFound => {
            return (Vec::new(), HashMap::new());
        }
        Err(error) => {
            tracing::debug!(
                event_name = "subagent_sidecar_dir_unreadable",
                session = %session_id,
                dir = %sidecar_dir.display(),
                %error,
                "subagent sidecars skipped: the directory could not be read",
            );
            return (Vec::new(), HashMap::new());
        }
    };
    let mut metas: Vec<(PathBuf, std::time::SystemTime)> = entries
        .filter_map(|entry| match entry {
            Ok(entry) => Some(entry),
            Err(error) => {
                tracing::debug!(
                    event_name = "subagent_sidecar_dir_entry_unreadable",
                    session = %session_id,
                    dir = %sidecar_dir.display(),
                    %error,
                    "a subagent sidecar entry was skipped: it could not be read",
                );
                None
            }
        })
        .filter(|entry| entry.file_name().to_str().is_some_and(|name| name.ends_with(".meta.json")))
        .filter_map(|entry| {
            let modified = match entry.metadata().and_then(|meta| meta.modified()) {
                Ok(modified) => modified,
                Err(error) => {
                    tracing::debug!(
                        event_name = "subagent_sidecar_meta_unreadable",
                        session = %session_id,
                        meta = %entry.path().display(),
                        %error,
                        "subagent sidecar meta skipped: its timestamp could not be read",
                    );
                    return None;
                }
            };
            Some((entry.path(), modified))
        })
        .collect();
    metas.sort_by(|left, right| right.1.cmp(&left.1).then_with(|| left.0.cmp(&right.0)));
    let mut budget = budget;
    let mut sidecars = Vec::new();
    let mut links = HashMap::new();
    for (meta, _) in metas {
        let Ok(body) = fs::read(&meta) else {
            tracing::debug!(
                event_name = "subagent_sidecar_meta_unreadable",
                session = %session_id,
                meta = %meta.display(),
                "subagent sidecar meta skipped: it could not be read",
            );
            continue;
        };
        let Ok(value) = serde_json::from_slice::<Value>(&body) else {
            tracing::debug!(
                event_name = "subagent_sidecar_meta_unreadable",
                session = %session_id,
                meta = %meta.display(),
                "subagent sidecar meta skipped: it is not JSON",
            );
            continue;
        };
        let Some(dispatch) =
            value.get("toolUseId").and_then(Value::as_str).filter(|id| !id.trim().is_empty())
        else {
            tracing::debug!(
                event_name = "subagent_sidecar_meta_unreadable",
                session = %session_id,
                meta = %meta.display(),
                "subagent sidecar meta skipped: it names no dispatch",
            );
            continue;
        };
        let Some(stem) = meta
            .file_name()
            .and_then(|name| name.to_str())
            .and_then(|name| name.strip_suffix(".meta.json"))
        else {
            tracing::debug!(
                event_name = "subagent_sidecar_meta_unreadable",
                session = %session_id,
                meta = %meta.display(),
                "subagent sidecar meta skipped: no sidecar name derives from it",
            );
            continue;
        };
        let agent = stem.strip_prefix("agent-").unwrap_or(stem);
        links.insert(
            agent.to_owned(),
            SidecarLink {
                dispatch: dispatch.to_owned(),
                parent_agent: value.get("parentAgentId").and_then(Value::as_str).map(str::to_owned),
            },
        );
        let sidecar = meta.with_file_name(format!("{stem}.jsonl"));
        let Ok(size) = fs::metadata(&sidecar).map(|info| info.len()) else {
            tracing::debug!(
                event_name = "subagent_sidecar_body_missing",
                session = %session_id,
                sidecar = %sidecar.display(),
                "subagent sidecar skipped: its body is missing",
            );
            continue;
        };
        if size > budget {
            tracing::debug!(
                event_name = "subagent_sidecar_budget_skipped",
                session = %session_id,
                sidecar = %sidecar.display(),
                "subagent sidecar skipped: the resume read's budget is spent",
            );
            continue;
        }
        budget -= size;
        let Ok(file) = fs::File::open(&sidecar) else {
            tracing::debug!(
                event_name = "subagent_sidecar_body_unreadable",
                session = %session_id,
                sidecar = %sidecar.display(),
                "subagent sidecar skipped: its body could not be opened",
            );
            continue;
        };
        // The fold stamps every row with its owner; the parse leaves the
        // wire's own value alone.
        let rows = parse_session_messages(file).messages;
        sidecars.push((dispatch.to_owned(), rows));
    }
    (sidecars, links)
}

/// The rows a transcript's messages convert to, typed as the wire sends them
/// and stamped with `session_id`.
///
/// **One conversion for every read of a transcript.** A resume hands these
/// messages over as history and a paging span read hands them over as a page
/// below the window; a second copy of the conversion is how the two would
/// come to type the same row differently.
pub(crate) fn messages_from_rows(
    rows: Vec<SessionMessage>,
    session_id: &str,
) -> Vec<forge_primitives::Message> {
    let raw: Vec<Value> = rows
        .into_iter()
        .map(|m| {
            let kind = match m.kind {
                forge_primitives::SessionMessageKind::User => "user",
                forge_primitives::SessionMessageKind::Assistant => "assistant",
                forge_primitives::SessionMessageKind::System => "system",
            };
            serde_json::json!({
                "type": kind,
                "uuid": m.uuid,
                "message": m.message,
                "parent_tool_use_id": m.parent_tool_use_id,
                "timestamp": m.timestamp,
                "tool_use_result": m.tool_use_result,
                "synthetic": m.synthetic,
            })
        })
        .collect();
    let mut messages = crate::replay::synthesize_replay_messages(&raw);
    // The synthesizer leaves the session empty so the caller picks the right
    // value: the resumed session for a resume, the transcript's own for a page.
    for message in &mut messages {
        match message {
            forge_primitives::Message::Assistant { session_id: s, .. }
            | forge_primitives::Message::User { session_id: s, .. }
            | forge_primitives::Message::StopHookSummary { session_id: s, .. }
            | forge_primitives::Message::CompactBoundary { session_id: s, .. }
            | forge_primitives::Message::PermissionDenied { session_id: s, .. }
            | forge_primitives::Message::Notification { session_id: s, .. } => {
                session_id.clone_into(s);
            }
            // The generic system rows the read keeps carry the same field,
            // optional because the CLI omits it on some subtypes.
            forge_primitives::Message::System { session_id: s, .. } => {
                *s = Some(session_id.to_owned());
            }
            _ => {}
        }
    }
    messages
}

/// The file a session id reads from: its project's directory when the caller
/// places it, the first copy named after it otherwise.
fn session_transcript(
    config_dir: &Path,
    session_id: &str,
    directory: Option<&str>,
) -> Option<PathBuf> {
    let file_name = format!("{session_id}.jsonl");
    if let Some(dir) = directory {
        let candidate = project_dir_for(config_dir, dir).join(&file_name);
        return candidate.is_file().then_some(candidate);
    }
    try_read_dir(&projects_dir_for(config_dir))?
        .flatten()
        .map(|entry| entry.path().join(&file_name))
        .find(|path| path.is_file())
}

/// How many bytes of a transcript one paging pass reads before growing, and
/// how wide a single window it takes may be.
///
/// **A page below the window's floor lies near the file's END**: the floor is
/// the oldest frame the window still holds, so the turns above it are the
/// newest rows of a file the session is still writing - some 4,000 rows, a
/// few hundred KB at the rows a transcript writes. The step is what a cold
/// session pays for its first paging request.
///
/// **The cap bounds one WINDOW, not the pass.** A read that comes up short
/// doubles and reads again, so a walk that reaches the cap has read the
/// windows under it as well - up to about twice this, and the anchor search
/// pays the same. What is capped is therefore how far a single window reaches,
/// and past it the transcript is treated as unreadable for paging rather than
/// read whole.
pub const SPAN_STEP_BYTES: u64 = 1024 * 1024;
pub const SPAN_CAP_BYTES: u64 = 64 * 1024 * 1024;

/// How many rows a page's own turn count may ask for.
///
/// Sized from the turns a page is cut into: the heaviest twenty-turn run
/// measured across the eight largest transcripts on the author's machine is
/// 2,799 rows, so 150 a turn asks for 3,000 at the twenty a subscribe carries,
/// which covers every page these transcripts would cut. A page whose turns
/// run heavier comes back short with a cursor to ask below with, rather than
/// converting a window's rows for turns nothing reads.
///
/// **The served hook rows ride inside that figure.** The SessionStart and
/// Setup runs a read now serves are about a seventh of a transcript's hook
/// rows and a few rows a session, so they do not move the 2,799 measurement;
/// a session that fired an unusual number of them is the "comes back short
/// with a cursor" case above, not a lost row.
pub const SPAN_ROWS_PER_TURN: usize = 150;

/// How many rows past a page's own count a cut may step back over, so that it
/// does not open on a result without its call.
///
/// A call and its results are adjacent rows in every transcript measured, so
/// one step is the ordinary case; the room is a short run of them - parallel
/// calls written as several result frames - and a run longer than this ends
/// the step rather than growing the page without bound.
const ROWS_PAST_CUT: usize = 8;

/// Whether a row carries a tool result: the kind of row a cut may not land on,
/// because the call it answers sits just above it in the page below.
fn carries_a_result(row: &SessionMessage) -> bool {
    row.kind == forge_primitives::SessionMessageKind::User
        && row.message.get("content").and_then(Value::as_array).is_some_and(|blocks| {
            blocks
                .iter()
                .any(|block| block.get("type").and_then(Value::as_str) == Some("tool_result"))
        })
}

/// Read the rows of `session_id`'s transcript that sit strictly before
/// `below`, newest first, up to `rows_wanted` of them.
///
/// **Row-level, on the file's own coordinates.** A page below a window's floor
/// is a running turn's older rows, and the conversation's frame numbering
/// cannot say where they are: a window whose front is a count cut inside a
/// running turn holds frames the CLI streamed and never wrote - thinking-token
/// frames among them - so a numbering that counted those frames cannot be
/// lined up against the file's rows with arithmetic. A row's own byte needs no
/// numbering, so the read works in bytes from an anchor: the row the page
/// above handed back as its cursor, or the file's own end when the walk's
/// first ask names none.
///
/// **Bounded by construction.** The read takes a window of the file, grows it
/// only while the rows it needs are not inside the window it took, and never
/// reads past [`SPAN_CAP_BYTES`]. `None` when the transcript cannot answer for
/// this session: no file, an anchor the file does not carry, an anchor
/// further from the file's end than the cap reaches, or a window that could
/// not be read. The caller answers all of them as the end of the history,
/// which is what a page this read cannot vouch for deserves.
pub fn read_rows(
    config_dir: &Path,
    session_id: &str,
    directory: Option<&str>,
    below: Option<&forge_primitives::TranscriptAnchor>,
    rows_wanted: usize,
) -> Option<forge_primitives::TranscriptSpan> {
    read_rows_with(
        config_dir,
        session_id,
        directory,
        below,
        rows_wanted,
        SPAN_STEP_BYTES,
        SPAN_CAP_BYTES,
    )
}

/// [`read_rows`] with the page's own budget in the caller's hands, so the
/// growth, the cap and the search bound can be driven by a test without a
/// transcript of the size the production budget is written for.
fn read_rows_with(
    config_dir: &Path,
    session_id: &str,
    directory: Option<&str>,
    below: Option<&forge_primitives::TranscriptAnchor>,
    rows_wanted: usize,
    step: u64,
    cap: u64,
) -> Option<forge_primitives::TranscriptSpan> {
    // The id names the file, so anything but a uuid is a path this refuses
    // rather than resolves.
    if !is_valid_uuid(session_id) {
        return None;
    }
    let path = session_transcript(config_dir, session_id, directory)?;
    let mut file = match fs::File::open(&path) {
        Ok(file) => file,
        Err(err) => {
            tracing::warn!(
                event_name = "span_read_open_failed",
                %session_id,
                path = %path.display(),
                %err,
                "the transcript could not be opened; the page is answered empty",
            );
            return None;
        }
    };
    let len = file.metadata().ok()?.len();
    if len == 0 {
        return None;
    }
    // **Where the read ends.** A named anchor places the window's end at its
    // own row, so the rows read are the ones strictly older than it; no
    // anchor means the file's end, whose rows are the newest the session
    // wrote - the walk's first ask on a window that holds no row to name.
    let end = match below {
        None => len,
        Some(anchor) => resolve_anchor(&mut file, session_id, len, anchor, step, cap)?,
    };

    // **The rows, walked backward from the end.** A window holds the rows just
    // below the byte it ends at; one that comes up short grows rather than
    // walking the file from its start, which on a long transcript is the whole
    // file. The windows meet on row boundaries, so no row is read twice.
    let mut kept: Vec<(SessionMessage, u64)> = Vec::new();
    let mut take = step.min(len).max(1);
    let mut at = end;
    let mut reached_start = false;
    let mut file_first: Option<u64> = None;
    loop {
        let window = match window_ending(&mut file, at, take) {
            Ok(window) => window,
            Err(err) => {
                tracing::warn!(
                    event_name = "span_read_failed",
                    %session_id,
                    from = at.saturating_sub(take),
                    %err,
                    "the transcript window could not be read; the page is answered empty",
                );
                return None;
            }
        };
        let rows: Vec<(SessionMessage, u64)> =
            window_rows(&window).into_iter().filter(|(_, offset)| *offset < at).collect();
        let next_at = rows.first().map(|(_, offset)| *offset);
        if window.from == 0 {
            reached_start = true;
            file_first = next_at;
        }
        // **The count, plus the room a cut steps over.** A cut may land among
        // result rows, and the rows it steps past are deferred to the page
        // below rather than dropped, so a little room past the count is read.
        let wanted = rows_wanted.saturating_add(ROWS_PAST_CUT);
        let room = wanted.saturating_sub(kept.len());
        let mut fresh = rows;
        if fresh.len() > room {
            // The ones nearest the boundary: a page is the newest of the older
            // rows, so what does not fit is left for the page below.
            fresh = fresh.split_off(fresh.len() - room);
        }
        let mut merged = fresh;
        merged.append(&mut kept);
        kept = merged;
        if kept.len() >= rows_wanted || window.from == 0 || take >= cap {
            break;
        }
        take = (take * 2).min(cap);
        at = next_at.unwrap_or(window.from);
        if at == 0 {
            break;
        }
    }

    // The page's own count: the room read past it was the cut's, and what the
    // cut did not need is not served.
    let over = kept.len().saturating_sub(rows_wanted);
    kept.drain(..over);
    // **A page never opens on a result without its call.** The span's oldest
    // row is where the page below stops, so a cut landing among result rows
    // would hand the fold a result whose call it does not carry - drawn with
    // its output nowhere. Those leading result rows are deferred instead: the
    // page below is cut from the call that produced them, which is older, and
    // serves them with it. Nothing is dropped - the row already read is simply
    // the page below's to carry - and a run longer than the room ends the step
    // rather than shortening the page without bound.
    let deferred = kept
        .iter()
        .position(|(row, _)| !carries_a_result(row))
        .map_or(0, |at| at.min(ROWS_PAST_CUT));
    kept.drain(..deferred);
    // The re-appended block, deduped the same way the whole-file read does it.
    // `kept` is oldest-first, so the original copy survives and the later one
    // is dropped.
    let mut seen = std::collections::HashSet::new();
    kept.retain(|(row, _)| !already_kept(&mut seen, &row.uuid));
    // The span runs to the history's start when the read reached the file's
    // own beginning AND the span's oldest row is the first row it found there:
    // a window that reached the start and held no row older than the boundary
    // has nothing left to serve either.
    let exhausted = reached_start
        && file_first.is_none_or(|first| kept.first().is_none_or(|(_, offset)| *offset == first));

    let mut messages = Vec::with_capacity(kept.len());
    let mut offsets = Vec::with_capacity(kept.len());
    for (row, offset) in kept {
        // **One row at a time**, so a row the synthesizer cannot decode is
        // dropped with its own byte rather than shifting every byte after it.
        if let Some(message) = messages_from_rows(vec![row], session_id).into_iter().next() {
            messages.push(message);
            offsets.push(offset);
        }
    }
    Some(forge_primitives::TranscriptSpan { messages, offsets, exhausted })
}

/// Where a read's window ends for a named anchor: the byte its row starts at,
/// seeking when the cursor carries a byte and searching from the file's end
/// when it does not.
///
/// `None` when the anchor cannot be placed - a row the file does not carry, or
/// one further from the end than [`SPAN_CAP_BYTES`] reaches - which the caller
/// answers as the end of the history.
fn resolve_anchor(
    file: &mut fs::File,
    session_id: &str,
    len: u64,
    anchor: &forge_primitives::TranscriptAnchor,
    step: u64,
    cap: u64,
) -> Option<u64> {
    if let Some(offset) = anchor.offset.filter(|offset| *offset <= len) {
        // The verify: a resumed or rewritten transcript carries something else
        // at the byte, and the search below is what answers that.
        if anchor.row.is_empty() || row_at(file, offset).is_some_and(|row| row.uuid == anchor.row) {
            return Some(offset);
        }
    }
    if anchor.row.is_empty() {
        return None;
    }
    let mut take = step.min(len).max(1);
    loop {
        let window = window_ending(file, len, take).ok()?;
        if let Some((_, offset)) =
            window_rows(&window).into_iter().find(|(row, _)| row.uuid == anchor.row)
        {
            return Some(offset);
        }
        if take >= cap || take >= len {
            // **Routine, not forge's own problem.** The frames the CLI
            // streams and never writes carry ids no row holds (a mid-turn
            // queued prompt reaches the file only as an attachment naming
            // it), so a miss is the caller's to walk past rather than a
            // failure here.
            tracing::debug!(
                event_name = "span_read_anchor_unreached",
                %session_id,
                row = %anchor.row,
                len,
                cap,
                "no row in the file names this anchor; the read answers the caller empty",
            );
            return None;
        }
        take = (take * 2).min(cap);
    }
}

/// One window of a transcript file: the bytes read and the byte they start
/// at.
struct Window {
    from: u64,
    bytes: Vec<u8>,
}

/// Read up to `take` bytes ending at `end`, and the byte before the window as
/// well.
///
/// **That one byte is what says whether the window starts on a row.** A window
/// beginning exactly at a row's first byte holds that whole row; one beginning
/// anywhere else holds a fragment of the row before it, and the two are
/// indistinguishable from inside the window.
fn window_ending(file: &mut fs::File, end: u64, take: u64) -> std::io::Result<Window> {
    let from = end.saturating_sub(take).saturating_sub(1);
    let mut bytes = vec![0_u8; usize::try_from(end - from).unwrap_or(0)];
    file.seek(SeekFrom::Start(from))?;
    file.read_exact(&mut bytes)?;
    Ok(Window { from, bytes })
}

/// The rows a window holds, each with the byte it starts at.
///
/// A window whose first byte follows anything but a newline begins inside a
/// row: everything before its next newline is that row's fragment - not a row,
/// and not counted. Every other line is whole.
fn window_rows(window: &Window) -> Vec<(SessionMessage, u64)> {
    // The window's first byte is the one before the window: a newline means
    // the window begins on a row, anything else means it begins inside one.
    let fragment = window.from > 0 && window.bytes.first() != Some(&b'\n');
    let mut rows = Vec::new();
    // A queued prompt is held until its remove row, whose byte is where the
    // row is served: the attachment's own byte is inside the turn the press
    // interrupted. A held row whose remove lies past the window never started
    // in this stretch, so it is dropped rather than served at the press.
    let mut queued = Vec::new();
    let mut at = window.from;
    for line in window.bytes.split(|byte| *byte == b'\n') {
        let start = at;
        at = at.saturating_add(line.len() as u64 + 1);
        if fragment && start == window.from {
            continue;
        }
        let Ok(text) = std::str::from_utf8(line) else { continue };
        let Ok(value) = serde_json::from_str::<Value>(text) else { continue };
        if let Some(remove) = queue_remove_row(&value) {
            rows.push((queued_start_row(&mut queued, &value, remove), start));
            continue;
        }
        let Some((row, _)) = transcript_row(&value) else { continue };
        if is_queued_attachment_row(&value) {
            queued.push(row);
            continue;
        }
        rows.push((row, start));
    }
    rows
}

/// The row that starts at `at`, for the anchored read's verify.
fn row_at(file: &mut fs::File, at: u64) -> Option<SessionMessage> {
    // One line, and a cap so a torn file cannot hand back the rest of itself.
    const LINE_CAP: u64 = 1024 * 1024;
    file.seek(SeekFrom::Start(at)).ok()?;
    let mut bytes = Vec::new();
    std::io::Read::take(&mut *file, LINE_CAP).read_to_end(&mut bytes).ok()?;
    let text = String::from_utf8_lossy(&bytes);
    session_row(text.lines().next().unwrap_or_default())
}

/// One line of a transcript as its row, or `None` for a line that is not one.
fn session_row(line: &str) -> Option<SessionMessage> {
    let line = line.trim_end_matches('\r');
    if line.is_empty() {
        return None;
    }
    let value = serde_json::from_str::<Value>(line).ok()?;
    // The queued prompt's row is served at this byte, so an anchor that names
    // it verifies here as well as in a window's row list.
    if let Some(row) = queue_remove_row(&value) {
        return Some(row);
    }
    transcript_row(&value).map(|(row, _)| row)
}

// ---------------------------------------------------------------------------
// Lite read - head + tail metadata extraction without full-file scan.
// ---------------------------------------------------------------------------

/// Head / tail snapshot of a session file - enough to recover all
/// [`SDKSessionInfo`] fields without a full scan. The `tag` field is
/// populated by a separate streaming line-scan because tag rows can
/// appear anywhere in the file (start-of-file at spawn, mid-file via
/// `/new` re-tag) and the head + tail windows miss both cases on
/// large transcripts.
struct LiteSessionFile {
    mtime: u64,
    size: u64,
    head: String,
    tail: String,
    tag: Option<String>,
}

/// Open a session file, stat it, read at most [`LITE_READ_BUF_SIZE`]
/// bytes from the head and the same from the tail. For files smaller
/// than the buffer, `tail == head` (single read). Returns `None` on
/// any I/O error or for empty files.
///
/// The tag scan that follows is NOT bounded by that buffer - it reads
/// the file end to end, so this is a lite read of the metadata and a
/// full read of the bytes. See [`scan_tag_from`].
///
/// Each `.ok()?` early return logs at debug level naming the step
/// that failed - without these, the session picker silently drops
/// sessions whose files have permission errors / are mid-truncation
/// / have a bad fd, which presents to the user as missing sessions
/// with no triage signal.
fn read_session_lite(
    path: &Path,
    cache: Option<&SessionTagCache>,
    scan_tag: bool,
) -> Option<LiteSessionFile> {
    let mut file = match fs::File::open(path) {
        Ok(f) => f,
        Err(e) => {
            tracing::debug!(target: crate::logging::targets::CATALOG_SCAN, path = %path.display(), error = %e, step = "open", "lite-read failed");
            return None;
        }
    };
    let meta = match file.metadata() {
        Ok(m) => m,
        Err(e) => {
            tracing::debug!(target: crate::logging::targets::CATALOG_SCAN, path = %path.display(), error = %e, step = "metadata", "lite-read failed");
            return None;
        }
    };
    let size = meta.len();
    if size == 0 {
        return None;
    }
    let mtime = meta
        .modified()
        .ok()
        .and_then(|t| t.duration_since(std::time::UNIX_EPOCH).ok())
        .map_or(0, |d| u64::try_from(d.as_millis()).unwrap_or(u64::MAX));

    let head_len = usize::try_from(LITE_READ_BUF_SIZE.min(size)).unwrap_or(usize::MAX);
    let mut head_bytes = vec![0u8; head_len];
    let read = match file.read(&mut head_bytes) {
        Ok(n) => n,
        Err(e) => {
            tracing::debug!(target: crate::logging::targets::CATALOG_SCAN, path = %path.display(), error = %e, step = "read_head", "lite-read failed");
            return None;
        }
    };
    head_bytes.truncate(read);
    if head_bytes.is_empty() {
        return None;
    }
    let head = String::from_utf8_lossy(&head_bytes).into_owned();

    let tail = if size <= LITE_READ_BUF_SIZE {
        head.clone()
    } else {
        let tail_offset = size - LITE_READ_BUF_SIZE;
        if let Err(e) = file.seek(SeekFrom::Start(tail_offset)) {
            tracing::debug!(target: crate::logging::targets::CATALOG_SCAN, path = %path.display(), error = %e, step = "seek_tail", "lite-read failed");
            return None;
        }
        let mut tail_bytes = vec![0u8; usize::try_from(LITE_READ_BUF_SIZE).unwrap_or(usize::MAX)];
        let read = match file.read(&mut tail_bytes) {
            Ok(n) => n,
            Err(e) => {
                tracing::debug!(target: crate::logging::targets::CATALOG_SCAN, path = %path.display(), error = %e, step = "read_tail", "lite-read failed");
                return None;
            }
        };
        tail_bytes.truncate(read);
        String::from_utf8_lossy(&tail_bytes).into_owned()
    };

    // The tag scan is the whole file. A caller that wants only the title
    // passes `false` and is not billed for the bytes that scan would read.
    let tag = if scan_tag {
        match tag_of(path, &mut file, size, cache) {
            Ok(tag) => tag,
            Err(e) => {
                tracing::debug!(target: crate::logging::targets::CATALOG_SCAN, path = %path.display(), error = %e, step = "seek_tag_scan", "lite-read tag-scan failed; resume will treat as untagged");
                None
            }
        }
    } else {
        None
    };

    Some(LiteSessionFile { mtime, size, head, tail, tag })
}

/// The last tag `path`'s transcript carries, with the read error
/// [`read_session_lite`] turns into "untagged" handed back as one: a
/// caller that has to tell an unreadable transcript from an untagged one
/// cannot use the lite read.
///
/// `size` is the file's length, `file` is positioned by this call, and
/// `cache` resumes the scan where the previous one stopped - a shrunk
/// file was truncated or replaced, so the whole thing is read again.
pub(crate) fn tag_of(
    path: &Path,
    file: &mut fs::File,
    size: u64,
    cache: Option<&SessionTagCache>,
) -> std::io::Result<Option<String>> {
    Ok(tag_and_project_of(path, file, size, cache)?.0)
}

/// The last tag AND the project its row names, with the same cache and
/// read-error behaviour as [`tag_of`]. A worker resume uses the project
/// to tell this project's session from another project's in a directory
/// both compose to.
pub(crate) fn tag_and_project_of(
    path: &Path,
    file: &mut fs::File,
    size: u64,
    cache: Option<&SessionTagCache>,
) -> std::io::Result<(Option<String>, Option<String>)> {
    let carried = cache
        .and_then(|c| c.get(path))
        .filter(|prior| prior.scanned_len <= size)
        .unwrap_or_default();
    file.seek(SeekFrom::Start(carried.scanned_len))?;
    let (tag, project, state) = scan_tag_from(BufReader::new(&mut *file), Some(carried));
    if let Some(cache) = cache {
        cache.put(path, state);
    }
    Ok((tag, project))
}

/// What a tag scan learned about one transcript, and how far into it the
/// answer is known to hold.
///
/// `scanned_len` is the offset just past the last newline the scan
/// consumed, NOT the file's length. Any trailing bytes with no newline
/// are a line the writer had not finished, so they are deliberately not
/// credited: the next scan re-reads that line from its true start. That
/// is what makes `scanned_len` a line boundary by construction, and a
/// line therefore cannot straddle it.
#[derive(Debug, Clone, Default, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct SessionTagScan {
    /// The last tag among the complete lines below `scanned_len`.
    pub tag: Option<String>,
    /// The project the last tag row names, when it carries one. Absent on
    /// a row written before the field existed, and on a cache entry
    /// persisted before it did.
    #[serde(default)]
    pub project: Option<String>,
    pub scanned_len: u64,
}

/// Carries the previous run's tag scans into this one and collects what
/// changed, so the whole catalog costs one store read up front and one
/// store write at the end rather than a transaction per transcript.
///
/// Owned by the caller because this crate has no store of its own: whoever
/// has the database fills it, passes it to [`list_sessions`], and persists
/// [`SessionTagCache::updates`] afterwards.
#[derive(Debug, Default)]
pub struct SessionTagCache {
    prior: std::collections::HashMap<String, SessionTagScan>,
    updated: std::sync::Mutex<Vec<(String, SessionTagScan)>>,
}

impl SessionTagCache {
    pub fn new(prior: std::collections::HashMap<String, SessionTagScan>) -> Self {
        Self { prior, updated: std::sync::Mutex::new(Vec::new()) }
    }

    fn get(&self, path: &Path) -> Option<SessionTagScan> {
        self.prior.get(path.to_str()?).cloned()
    }

    /// Whether this transcript carried `wanted` at its last scan, when the
    /// file has not changed since. `None` when the tag is not known - nothing
    /// is cached, or the entry is behind the file on disk, where a `/new`
    /// re-tag can land - so the caller reads the file rather than deciding.
    fn carries_tag(&self, path: &Path, wanted: &str) -> Option<bool> {
        // The cache first: a path it has never seen costs no metadata call on
        // a cold listing.
        let held = self.get(path)?;
        let size = std::fs::metadata(path).ok()?.len();
        (held.scanned_len == size).then(|| held.tag.as_deref() == Some(wanted))
    }

    fn put(&self, path: &Path, scan: SessionTagScan) {
        let Some(key) = path.to_str() else { return };
        // Unchanged files re-derive the entry they already had; writing it
        // back would rewrite the whole table on every boot.
        if self.prior.get(key) == Some(&scan) {
            return;
        }
        if let Ok(mut updated) = self.updated.lock() {
            updated.push((key.to_owned(), scan));
        }
    }

    /// Entries whose scan moved this run, for the caller to persist.
    pub fn updates(&self) -> Vec<(String, SessionTagScan)> {
        self.updated.lock().map(|u| u.clone()).unwrap_or_default()
    }
}

/// Return the value of the LAST `{"type":"tag"}` row's `"tag"` field.
/// Empty strings are filtered. `None` when no tag row is present or any
/// read errors out.
///
/// Last-wins semantics: a `/new` re-tag appended later in the
/// transcript supersedes the original spawn tag.
///
/// Reads into one reused buffer rather than over `BufRead::lines`,
/// whose per-line `String` is pure overhead here: this runs over every
/// byte of every transcript in the catalog on the boot path, and the
/// tag rows it is looking for are a handful of lines in gigabytes.
///
/// `carried` is the answer a previous scan reached
/// at its own `scanned_len`, which stands unless these bytes hold a
/// later tag - last-wins makes that sound, since every line here follows
/// every line the carried answer was computed over.
///
/// Returns the tag INCLUDING any trailing unterminated line, which is
/// what a whole-file scan has always reported, alongside the state to
/// cache, which excludes it.
fn scan_tag_from<R: BufRead>(
    mut reader: R,
    carried: Option<SessionTagScan>,
) -> (Option<String>, Option<String>, SessionTagScan) {
    let mut state = carried.unwrap_or_default();
    // Tracked apart from `state.tag` so a tag on a half-written line is
    // reported now but not cached: the line is re-read next time.
    let mut last_tag = state.tag.clone();
    let mut last_project = state.project.clone();
    let mut line = Vec::new();
    loop {
        line.clear();
        match reader.read_until(b'\n', &mut line) {
            Ok(0) => break,
            Ok(_) => {}
            Err(e) => {
                tracing::debug!(target: crate::logging::targets::CATALOG_SCAN, error = %e, step = "tag_scan_line", "lite-read tag-scan line read failed; ending scan with last seen tag");
                break;
            }
        }
        let complete = line.last() == Some(&b'\n');
        let raw_len = line.len() as u64;
        // `lines()` yields neither terminator, and both are stripped
        // before the prefix test so a CRLF transcript compares the same
        // bytes it always did.
        if complete {
            line.pop();
        }
        if line.last() == Some(&b'\r') {
            line.pop();
        }
        // Every line is still validated, not only the candidates: a
        // transcript that is not UTF-8 ended the scan before, and where
        // it ends decides which tag wins. Breaking before the line is
        // credited leaves it to be retried rather than skipped for good.
        let text = match std::str::from_utf8(&line) {
            Ok(text) => text,
            Err(e) => {
                tracing::debug!(target: crate::logging::targets::CATALOG_SCAN, error = %e, step = "tag_scan_line", "lite-read tag-scan line read failed; ending scan with last seen tag");
                break;
            }
        };
        if text.starts_with("{\"type\":\"tag\"")
            && let Some(tag) = extract_last_json_string_field(text, "tag")
            && !tag.is_empty()
        {
            last_tag = Some(tag);
            last_project = extract_json_string_field(text, "project");
            if complete {
                state.tag.clone_from(&last_tag);
                state.project.clone_from(&last_project);
            }
        }
        if complete {
            state.scanned_len += raw_len;
        }
    }
    (last_tag, last_project, state)
}

/// Find the first byte offset where `needle` begins in `haystack`.
fn find_bytes(haystack: &[u8], needle: &[u8]) -> Option<usize> {
    if needle.is_empty() || needle.len() > haystack.len() {
        return None;
    }
    haystack.windows(needle.len()).position(|w| w == needle)
}

/// Extract the first occurrence of a JSON string field (`"key":"value"`
/// or `"key": "value"`). Scans bytes directly to survive partial tail
/// reads; unescapes via `serde_json` only when the value contains a
/// backslash. Returns `None` when the field is absent or unterminated.
fn extract_json_string_field(text: &str, key: &str) -> Option<String> {
    let bytes = text.as_bytes();
    let compact = format!("\"{key}\":\"");
    let spaced = format!("\"{key}\": \"");
    for pattern in [compact.as_bytes(), spaced.as_bytes()] {
        if let Some(idx) = find_bytes(bytes, pattern) {
            let value_start = idx + pattern.len();
            let mut i = value_start;
            while i < bytes.len() {
                if bytes[i] == b'\\' {
                    i += 2;
                    continue;
                }
                if bytes[i] == b'"' {
                    let raw = std::str::from_utf8(&bytes[value_start..i]).ok()?;
                    return Some(unescape_json_string(raw));
                }
                i += 1;
            }
        }
    }
    None
}

/// Like [`extract_json_string_field`] but returns the LAST occurrence.
fn extract_last_json_string_field(text: &str, key: &str) -> Option<String> {
    let bytes = text.as_bytes();
    let compact = format!("\"{key}\":\"");
    let spaced = format!("\"{key}\": \"");
    let mut last: Option<String> = None;
    // Track the byte offset of the winning match: the compact and
    // spaced patterns are scanned separately, so without comparing
    // positions the spaced scan would clobber a later compact match.
    let mut last_pos: Option<usize> = None;
    for pattern in [compact.as_bytes(), spaced.as_bytes()] {
        let mut search_from = 0usize;
        while search_from < bytes.len() {
            let remaining = &bytes[search_from..];
            let Some(rel_idx) = find_bytes(remaining, pattern) else {
                break;
            };
            let idx = search_from + rel_idx;
            let value_start = idx + pattern.len();
            let mut i = value_start;
            while i < bytes.len() {
                if bytes[i] == b'\\' {
                    i += 2;
                    continue;
                }
                if bytes[i] == b'"' {
                    if last_pos.is_none_or(|p| idx >= p)
                        && let Ok(raw) = std::str::from_utf8(&bytes[value_start..i])
                    {
                        last = Some(unescape_json_string(raw));
                        last_pos = Some(idx);
                    }
                    break;
                }
                i += 1;
            }
            search_from = i + 1;
        }
    }
    last
}

/// Unescape a JSON string value. No-op when there are no backslashes.
fn unescape_json_string(raw: &str) -> String {
    if !raw.contains('\\') {
        return raw.to_string();
    }
    let wrapped = format!("\"{raw}\"");
    serde_json::from_str::<String>(&wrapped).unwrap_or_else(|_| raw.to_string())
}

/// Extract the first meaningful user prompt from a JSONL head chunk.
/// Skips `tool_result`, `isMeta`, `isCompactSummary`, slash-command
/// messages (with command-name fallback), and the fixed-prefix skip
/// patterns the CLI's `_SKIP_FIRST_PROMPT_PATTERN` matches. Truncates to
/// 200 chars with an ellipsis.
fn extract_first_prompt_from_head(head: &str) -> Option<String> {
    let mut command_fallback: Option<String> = None;
    for line in head.split('\n') {
        if !line.contains("\"type\":\"user\"") && !line.contains("\"type\": \"user\"") {
            continue;
        }
        if line.contains("\"tool_result\"") {
            continue;
        }
        if line.contains("\"isMeta\":true") || line.contains("\"isMeta\": true") {
            continue;
        }
        if line.contains("\"isCompactSummary\":true") || line.contains("\"isCompactSummary\": true")
        {
            continue;
        }
        let Ok(entry) = serde_json::from_str::<Value>(line) else {
            continue;
        };
        if entry.get("type").and_then(Value::as_str) != Some("user") {
            continue;
        }
        let Some(message) = entry.get("message") else {
            continue;
        };
        let Some(content) = message.get("content") else {
            continue;
        };
        let texts: Vec<String> = if let Some(s) = content.as_str() {
            vec![s.to_string()]
        } else if let Some(arr) = content.as_array() {
            arr.iter()
                .filter_map(|b| {
                    (b.get("type").and_then(Value::as_str) == Some("text"))
                        .then(|| b.get("text").and_then(Value::as_str).map(str::to_string))
                        .flatten()
                })
                .collect()
        } else {
            Vec::new()
        };
        for raw in texts {
            let result = raw.replace('\n', " ");
            let result = result.trim();
            if result.is_empty() {
                continue;
            }
            if let Some(cmd) = extract_command_name(result) {
                if command_fallback.is_none() {
                    command_fallback = Some(cmd);
                }
                continue;
            }
            if should_skip_first_prompt(result) {
                continue;
            }
            let truncated = if result.chars().count() > 200 {
                let mut buf: String = result.chars().take(200).collect();
                while buf.ends_with(char::is_whitespace) {
                    buf.pop();
                }
                buf.push('\u{2026}');
                buf
            } else {
                result.to_string()
            };
            return Some(truncated);
        }
    }
    command_fallback
}

/// Extract `<command-name>CMD</command-name>` when present.
pub(crate) fn extract_command_name(s: &str) -> Option<String> {
    const OPEN: &str = "<command-name>";
    const CLOSE: &str = "</command-name>";
    let open = s.find(OPEN)?;
    let after = &s[open + OPEN.len()..];
    let close = after.find(CLOSE)?;
    Some(after[..close].to_string())
}

/// Fixed-prefix counterpart to the CLI's `_SKIP_FIRST_PROMPT_PATTERN`.
pub(crate) fn should_skip_first_prompt(s: &str) -> bool {
    const PREFIXES: [&str; 4] =
        ["<local-command-stdout>", "<session-start-hook>", "<tick>", "<goal>"];
    if PREFIXES.iter().any(|p| s.starts_with(p)) {
        return true;
    }
    if s.starts_with("[Request interrupted by user") && s.contains(']') {
        return true;
    }
    let trimmed = s.trim();
    for (open, close) in
        [("<ide_opened_file>", "</ide_opened_file>"), ("<ide_selection>", "</ide_selection>")]
    {
        if trimmed.starts_with(open) && trimmed.ends_with(close) {
            return true;
        }
    }
    false
}

/// The one transcript at `path`, read for its title and clock.
///
/// A row the catalog mirrored at `Connected` carries a placeholder title
/// until the next scan reads the transcript; this reads that one file for a
/// caller that needs the title now, and `None` when there is no readable
/// transcript there.
///
/// **The tag is not read**, and is `None` on the answer: a tag scan covers
/// the whole file, and a caller that only needs the title must not pay for
/// bytes it does not use.
pub fn read_session_info_at(path: &Path) -> Option<SDKSessionInfo> {
    read_session_info_with(path, None, false)
}

fn read_session_info(path: &Path, cache: Option<&SessionTagCache>) -> Option<SDKSessionInfo> {
    read_session_info_with(path, cache, true)
}

fn read_session_info_with(
    path: &Path,
    cache: Option<&SessionTagCache>,
    scan_tag: bool,
) -> Option<SDKSessionInfo> {
    let session_id = path.file_stem().and_then(|s| s.to_str())?.to_string();
    let lite = read_session_lite(path, cache, scan_tag)?;
    let storage_key = path
        .parent()
        .and_then(Path::file_name)
        .and_then(|name| name.to_str())
        .unwrap_or_default()
        .to_owned();
    let mut info = parse_session_info_from_lite(&session_id, &lite, None)?;
    info.storage_key = storage_key;
    Some(info)
}

/// Build an [`SDKSessionInfo`] from a lite head/tail read. Skips
/// sidechain transcripts and metadata-only sessions (no summary
/// after all fallbacks).
fn parse_session_info_from_lite(
    session_id: &str,
    lite: &LiteSessionFile,
    project_path: Option<&str>,
) -> Option<SDKSessionInfo> {
    let head = lite.head.as_str();
    let tail = lite.tail.as_str();

    let first_line = head.find('\n').map_or(head, |idx| &head[..idx]);
    if first_line.contains("\"isSidechain\":true") || first_line.contains("\"isSidechain\": true") {
        return None;
    }

    let custom_title = extract_last_json_string_field(tail, "customTitle")
        .or_else(|| extract_last_json_string_field(head, "customTitle"))
        .or_else(|| extract_last_json_string_field(tail, "aiTitle"))
        .or_else(|| extract_last_json_string_field(head, "aiTitle"));
    let first_prompt = extract_first_prompt_from_head(head);
    // Bias toward labels that identify what the session is about, not
    // what was last said in it. customTitle / aiTitle (claude-written
    // 4-6 word title) is best; first_prompt (the user's opening
    // prompt) is the next-best identifier; lastPrompt (a mid-session
    // follow-up that needs context to interpret) is the fallback
    // because it produces unreadable labels at narrow widths.
    let summary = custom_title
        .clone()
        .or_else(|| first_prompt.clone())
        .or_else(|| extract_last_json_string_field(tail, "lastPrompt"))
        .or_else(|| extract_last_json_string_field(tail, "summary"))?;

    let git_branch = extract_last_json_string_field(tail, "gitBranch")
        .or_else(|| extract_json_string_field(head, "gitBranch"));
    let cwd = extract_json_string_field(head, "cwd").or_else(|| project_path.map(str::to_string));
    // Tag is pre-extracted by `read_session_lite` via a full-file
    // line-scan; neither head nor tail alone covers every position
    // a `{"type":"tag"}` row can land at.
    let tag = lite.tag.clone();
    let created_at = extract_json_string_field(first_line, "timestamp")
        .and_then(|ts| parse_rfc3339_ms(&ts).ok());

    Some(SDKSessionInfo {
        session_id: session_id.to_string(),
        summary,
        last_modified: lite.mtime,
        file_size: Some(lite.size),
        custom_title,
        first_prompt,
        git_branch,
        cwd,
        storage_key: String::new(),
        tag,
        created_at,
    })
}

/// Parse the CLI's RFC-3339 timestamp (e.g. `2026-04-22T04:15:27.123Z`)
/// into Unix epoch milliseconds. Sub-millisecond precision is truncated.
pub(crate) fn parse_rfc3339_ms(ts: &str) -> Result<u64, time::error::Parse> {
    let dt = time::OffsetDateTime::parse(ts, &time::format_description::well_known::Rfc3339)?;
    let nanos = dt.unix_timestamp_nanos();
    Ok(u64::try_from(nanos / 1_000_000).unwrap_or(0))
}

#[cfg(test)]
mod tests {

    use super::*;

    /// Buffer tracing output so an emitted record can be read back.
    #[derive(Clone, Default)]
    struct LogCapture(std::sync::Arc<parking_lot::Mutex<Vec<u8>>>);

    impl std::io::Write for LogCapture {
        fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
            self.0.lock().extend_from_slice(buf);
            Ok(buf.len())
        }
        fn flush(&mut self) -> std::io::Result<()> {
            Ok(())
        }
    }

    impl<'a> tracing_subscriber::fmt::MakeWriter<'a> for LogCapture {
        type Writer = Self;
        fn make_writer(&'a self) -> Self::Writer {
            self.clone()
        }
    }

    fn capture_logs(f: impl FnOnce()) -> String {
        let capture = LogCapture::default();
        let subscriber = tracing_subscriber::fmt().with_writer(capture.clone()).finish();
        tracing::subscriber::with_default(subscriber, f);
        let bytes = capture.0.lock().clone();
        String::from_utf8_lossy(&bytes).into_owned()
    }

    /// A renamed disk field is the one drift this scan cannot otherwise see:
    /// the count stays right and the row still draws, bare or half-filled. The
    /// live arm warns on the same degradation, so the read path has to as well
    /// or the only trace of it is a row with no count and nothing saying why.
    ///
    /// **The shapes are the same failure and are keyed together**: the outer
    /// key renamed, any one of the three fields inside it renamed - the
    /// pre-count's is the one the primitives test calls plausible, and the
    /// trigger's and the post-count's are the symmetric cases - and no metadata
    /// at all. Each names what it left out, so the record points at the key
    /// that moved rather than at the row.
    #[test]
    fn a_boundary_row_whose_facts_did_not_survive_is_logged() {
        for (shape, row, missing) in [
            (
                "outer key renamed",
                r#"{"type":"system","subtype":"compact_boundary","uuid":"cb1","session_id":"s1","compactMetadataMoved":{"trigger":"auto","preTokens":41207}}"#,
                "trigger, preTokens, postTokens",
            ),
            (
                "count renamed inside it",
                r#"{"type":"system","subtype":"compact_boundary","uuid":"cb1","session_id":"s1","compactMetadata":{"trigger":"auto","pre_tokens":41207}}"#,
                "preTokens, postTokens",
            ),
            (
                "trigger renamed inside it",
                r#"{"type":"system","subtype":"compact_boundary","uuid":"cb1","session_id":"s1","compactMetadata":{"triggers":"auto","preTokens":41207}}"#,
                "trigger, postTokens",
            ),
            (
                "post count renamed inside it",
                r#"{"type":"system","subtype":"compact_boundary","uuid":"cb1","session_id":"s1","compactMetadata":{"trigger":"auto","preTokens":41207,"post_tokens":1265}}"#,
                "postTokens",
            ),
            (
                "no metadata at all",
                r#"{"type":"system","subtype":"compact_boundary","uuid":"cb1","session_id":"s1"}"#,
                "trigger, preTokens, postTokens",
            ),
        ] {
            let log = capture_logs(|| {
                parse_session_messages(row.as_bytes());
            });

            assert!(log.contains("cb1"), "the record for a {shape} names its row: {log}");
            assert!(log.contains("s1"), "and the session it was read for: {log}");
            assert!(log.contains(missing), "and the fact {shape} left out: {log}");
        }
    }

    /// The negative control: a row carrying its metadata warns about nothing,
    /// so the record above is about the drift and not about every boundary.
    #[test]
    fn a_boundary_row_with_its_metadata_logs_nothing() {
        let log = capture_logs(|| {
            parse_session_messages(
                br#"{"type":"system","subtype":"compact_boundary","uuid":"cb1","session_id":"s1","compactMetadata":{"trigger":"auto","preTokens":41207,"postTokens":1265}}"#
                    .as_slice(),
            );
        });

        assert!(log.is_empty(), "a row the scan can read warns about nothing: {log}");
    }

    #[test]
    fn sanitize_ascii_only_passthrough() {
        assert_eq!(sanitize_path("alphanum123"), "alphanum123");
    }

    #[test]
    fn sanitize_replaces_non_alphanum_with_hyphens() {
        assert_eq!(
            sanitize_path("/Users/developer/projects/forge"),
            "-Users-developer-projects-forge"
        );
    }

    #[test]
    fn extract_last_json_string_field_picks_globally_last_across_forms() {
        // The compact form appears LATER than the spaced form; the
        // globally-last value must win regardless of which pattern is
        // scanned first (the bug was the spaced scan clobbering a later
        // compact match).
        let text = r#"{"tag": "early-spaced"} {"tag":"late-compact"}"#;
        assert_eq!(extract_last_json_string_field(text, "tag"), Some("late-compact".to_owned()));
        // Reverse: spaced later than compact.
        let text2 = r#"{"tag":"early-compact"} {"tag": "late-spaced"}"#;
        assert_eq!(extract_last_json_string_field(text2, "tag"), Some("late-spaced".to_owned()));
    }

    #[test]
    fn simple_hash_matches_known_value() {
        // Reference: _simple_hash("foo") → "26di" (computed from the
        // same 32-bit JS-style hash algorithm).
        assert_eq!(simple_hash("foo"), "26di");
    }

    #[test]
    fn long_path_gets_hash_suffix() {
        let long = "a".repeat(300);
        let got = sanitize_path(&long);
        assert_eq!(got.len(), MAX_SANITIZED_LENGTH + 1 + simple_hash(&long).len());
    }

    #[test]
    fn uuid_validator_accepts_canonical() {
        assert!(is_valid_uuid("550e8400-e29b-41d4-a716-446655440000"));
    }

    #[test]
    fn uuid_validator_rejects_garbage() {
        assert!(!is_valid_uuid("not-a-uuid"));
        assert!(!is_valid_uuid("550e8400e29b41d4a716446655440000"));
        assert!(!is_valid_uuid(""));
    }

    #[test]
    fn iso_parser_handles_millis() {
        let ms = parse_rfc3339_ms("2026-04-22T00:00:00.500Z").unwrap();
        assert_eq!(ms % 1000, 500);
    }

    #[test]
    fn extract_json_string_field_finds_compact_form() {
        let t = r#"noise {"type":"user","message":{"content":"hi"}} noise"#;
        assert_eq!(extract_json_string_field(t, "content"), Some("hi".to_string()));
    }

    #[test]
    fn extract_json_string_field_finds_spaced_form() {
        let t = r#"{"gitBranch": "main"}"#;
        assert_eq!(extract_json_string_field(t, "gitBranch"), Some("main".to_string()));
    }

    #[test]
    fn extract_json_string_field_handles_escaped_quotes() {
        let t = r#"{"customTitle":"he said \"hi\""}"#;
        assert_eq!(
            extract_json_string_field(t, "customTitle"),
            Some(r#"he said "hi""#.to_string())
        );
    }

    #[test]
    fn extract_last_json_string_field_picks_last() {
        let t = r#"{"tag":"old"} {"tag":"new"}"#;
        assert_eq!(extract_last_json_string_field(t, "tag"), Some("new".to_string()));
    }

    #[test]
    fn first_prompt_skips_local_command_stdout() {
        let head = r#"{"type":"user","message":{"content":"<local-command-stdout>out</local-command-stdout>"}}
{"type":"user","message":{"content":"actual prompt"}}"#;
        assert_eq!(extract_first_prompt_from_head(head), Some("actual prompt".to_string()));
    }

    #[test]
    fn first_prompt_falls_back_to_command_name() {
        let head = r#"{"type":"user","message":{"content":"<command-name>foo</command-name>"}}"#;
        assert_eq!(extract_first_prompt_from_head(head), Some("foo".to_string()));
    }

    #[test]
    fn first_prompt_skips_tool_result_line() {
        let head = r#"{"type":"user","message":{"content":[{"type":"tool_result","content":"x"}]}}
{"type":"user","message":{"content":"real prompt"}}"#;
        assert_eq!(extract_first_prompt_from_head(head), Some("real prompt".to_string()));
    }

    #[test]
    fn parse_session_info_skips_sidechain() {
        let head = "{\"isSidechain\":true,\"type\":\"user\"}\n".to_string();
        let lite = LiteSessionFile { mtime: 0, size: 1, head: head.clone(), tail: head, tag: None };
        assert!(parse_session_info_from_lite("abc", &lite, None).is_none());
    }

    #[test]
    fn parse_session_info_skips_metadata_only() {
        let content = "{\"type\":\"tag\",\"tag\":\"meta\"}\n".to_string();
        let lite = LiteSessionFile {
            mtime: 10,
            size: content.len() as u64,
            head: content.clone(),
            tail: content,
            tag: Some("meta".to_string()),
        };
        // No custom_title, no aiTitle, no lastPrompt, no summary,
        // no first_prompt → skipped.
        assert!(parse_session_info_from_lite("abc", &lite, None).is_none());
    }

    #[test]
    fn parse_session_info_extracts_prompt_and_tag() {
        let content = r#"{"type":"user","timestamp":"2026-04-22T00:00:00.000Z","gitBranch":"main","cwd":"/p","message":{"content":"hello"}}
{"type":"tag","tag":"mytag"}
"#
        .to_string();
        let lite = LiteSessionFile {
            mtime: 99,
            size: content.len() as u64,
            head: content.clone(),
            tail: content,
            tag: Some("mytag".to_string()),
        };
        let info = parse_session_info_from_lite("abc", &lite, None).expect("some");
        assert_eq!(info.first_prompt.as_deref(), Some("hello"));
        assert_eq!(info.summary, "hello");
        assert_eq!(info.tag.as_deref(), Some("mytag"));
        assert_eq!(info.git_branch.as_deref(), Some("main"));
        assert_eq!(info.cwd.as_deref(), Some("/p"));
        assert!(info.created_at.is_some());
    }

    #[test]
    fn find_session_tag_ignores_tag_on_tool_use_lines() {
        // A git-tag tool_use shouldn't be picked up as a session tag -
        // the `"tag"` string appears but the line isn't `{"type":"tag"`.
        let content = r#"{"type":"user","message":{"content":"hi"}}
{"type":"assistant","message":{"content":[{"type":"tool_use","input":{"command":"git tag","tag":"v1.0"}}]}}
"#;
        assert_eq!(scan_tag_from(std::io::Cursor::new(content), None).0, None);
    }

    #[test]
    fn parse_session_info_prefers_custom_title_over_last_prompt() {
        let content = r#"{"type":"user","message":{"content":"initial"}}
{"customTitle":"Curated","lastPrompt":"last"}
"#
        .to_string();
        let lite = LiteSessionFile {
            mtime: 0,
            size: content.len() as u64,
            head: content.clone(),
            tail: content,
            tag: None,
        };
        let info = parse_session_info_from_lite("abc", &lite, None).expect("some");
        assert_eq!(info.summary, "Curated");
        assert_eq!(info.custom_title.as_deref(), Some("Curated"));
    }

    /// **A queued prompt's row sits where the CLI took it, not where it was
    /// enqueued.** The attachment is written at the press, inside the turn it
    /// interrupted; the `queue-operation remove` row is written when the CLI
    /// took the prompt. The live path draws the row at that start, so a read
    /// that served the attachment's position put the words above work written
    /// after the press (rule 25). An attachment with no remove never started
    /// and draws nothing, exactly as the live path drew nothing.
    #[test]
    fn parse_session_messages_places_a_queued_prompt_at_its_start() {
        // claude persists mid-turn queued inputs as
        // `{"type":"attachment", "attachment":{"type":"queued_command",
        // "prompt":"...", "commandMode":"prompt"}}`, and the start as a
        // `queue-operation remove` naming the same words.
        let jsonl = r#"{"type":"user","message":{"role":"user","content":"plain prompt"},"uuid":"u1","session_id":"s1","timestamp":"2026-11-03T12:00:50.000Z"}
{"type":"attachment","attachment":{"type":"queued_command","prompt":"queued prompt","commandMode":"prompt","source_uuid":"p1"},"uuid":"a1","session_id":"s1","timestamp":"2026-11-03T12:00:58.000Z"}
{"type":"assistant","message":{"role":"assistant","content":[{"type":"text","text":"after the press"}]},"uuid":"as1","session_id":"s1","timestamp":"2026-11-03T12:00:59.000Z"}
{"type":"queue-operation","operation":"remove","timestamp":"2026-11-03T12:01:10.000Z","sessionId":"s1","content":"queued prompt"}
{"type":"assistant","message":{"role":"assistant","content":[{"type":"text","text":"the answer"}]},"uuid":"as2","session_id":"s1","timestamp":"2026-11-03T12:01:11.000Z"}
{"type":"attachment","attachment":{"type":"queued_command","prompt":"never started","commandMode":"prompt"},"uuid":"a2","session_id":"s1","timestamp":"2026-11-03T12:02:00.000Z"}
"#;
        let msgs = parse_session_messages(jsonl.as_bytes()).messages;
        assert_eq!(msgs.len(), 4, "the prompt that never started draws no row");
        assert_eq!(msgs[0].uuid, "u1");
        assert_eq!(msgs[1].uuid, "as1");
        assert_eq!(msgs[3].uuid, "as2");
        assert!(
            msgs[2].uuid.starts_with("queue-remove-2026-11-03T12:01:10.000Z-"),
            "the row is named by the start it was served at: {:?}",
            msgs[2].uuid,
        );
        let content = msgs[2]
            .message
            .get("content")
            .and_then(Value::as_array)
            .expect("synthesised message has content array");
        assert_eq!(content.len(), 1);
        assert_eq!(content[0].get("type").and_then(Value::as_str), Some("queued_command"));
        assert_eq!(content[0].get("prompt").and_then(Value::as_str), Some("queued prompt"));
        assert_eq!(content[0].get("commandMode").and_then(Value::as_str), Some("prompt"));
        assert_eq!(
            content[0].get("source_uuid").and_then(Value::as_str),
            Some("p1"),
            "the prompt id the live row pairs with",
        );
        assert_eq!(
            msgs[2].timestamp.as_deref(),
            Some("2026-11-03T12:01:10.000Z"),
            "the start's clock, which is where the row is placed",
        );
    }

    /// **A hook run the file persisted is a row the live chat drew.** The live
    /// stream sends each run as a `system/hook_response` frame; the transcript
    /// keeps it as an attachment. Dropping the row left a server page with
    /// none of the SessionStart rows the reader was looking at (rule 25), and
    /// every landing that merged against such a page moved the live rows it
    /// did not carry.
    ///
    /// **The served set is the live stream's, not the file's.** The CLI gates
    /// live hook frames on SessionStart and Setup, so only those events are
    /// served; `hook_additional_context` is aggregate context, not a run, and
    /// the live path drew it as no row at all.
    #[test]
    fn parse_session_messages_serves_the_hook_runs_the_live_wire_forwards() {
        let jsonl = r###"{"type":"attachment","attachment":{"type":"hook_success","hookName":"SessionStart:startup","hookEvent":"SessionStart","content":"## The PR review gate","stdout":"## The PR review gate","exitCode":0,"toolUseID":"tool-1"},"uuid":"h1","sessionId":"s1"}
{"type":"attachment","attachment":{"type":"hook_success","hookName":"PreToolUse:Skill","hookEvent":"PreToolUse","content":"not served live","stdout":"not served live","exitCode":0,"toolUseID":"tool-1"},"uuid":"h-skip","sessionId":"s1"}
{"type":"attachment","attachment":{"type":"hook_additional_context","content":["aggregate context"],"hookName":"SessionStart:startup","hookEvent":"SessionStart","toolUseID":"tool-1"},"uuid":"h-context","sessionId":"s1"}
{"type":"assistant","message":{"id":"m1","role":"assistant","model":"claude-opus-5","content":[{"type":"text","text":"after the first run"}]},"uuid":"a1","sessionId":"s1"}
{"type":"attachment","attachment":{"type":"hook_success","hookName":"Setup:init","hookEvent":"Setup","content":["line one","line two"],"stdout":"line one\nline two","exitCode":0,"toolUseID":"tool-1"},"uuid":"h2","sessionId":"s1"}
{"type":"attachment","attachment":{"type":"hook_non_blocking_error","hookName":"SessionStart:startup","hookEvent":"SessionStart","stdout":"","stderr":"Failed with non-blocking status code","exitCode":141,"command":"bash til.sh","durationMs":45,"toolUseID":"tool-1"},"uuid":"h3","sessionId":"s1"}
{"type":"attachment","attachment":{"type":"hook_cancelled","hookName":"SessionStart:startup","hookEvent":"SessionStart","command":"check.sh","durationMs":3018,"timedOut":true,"timeoutMs":3000,"toolUseID":"tool-1"},"uuid":"h4","sessionId":"s1"}
{"type":"attachment","attachment":{"type":"hook_success","hookName":"SessionStart:startup","hookEvent":"SessionStart","content":"","stdout":"plugin lines","stderr":"","exitCode":0,"toolUseID":"tool-1"},"uuid":"h5","sessionId":"s1"}
"###;
        let msgs = parse_session_messages(jsonl.as_bytes()).messages;
        assert_eq!(
            msgs.iter().map(|m| m.uuid.clone()).collect::<Vec<_>>().join(","),
            "h1,a1,h2,h3,h4,h5",
            "the served runs, in file order, with the assistant row between them"
        );
        assert!(
            msgs.iter().filter(|m| !matches!(m.kind, SessionMessageKind::System)).count() == 1,
            "a hook row is a system frame, the shape the live path draws"
        );

        let h1 = &msgs[0].message;
        assert_eq!(h1.get("subtype").and_then(Value::as_str), Some("hook_response"));
        assert_eq!(h1.get("hook_id").and_then(Value::as_str), Some("h1"));
        assert_eq!(h1.get("uuid").and_then(Value::as_str), Some("h1"));
        assert_eq!(h1.get("hook_name").and_then(Value::as_str), Some("SessionStart:startup"));
        assert_eq!(h1.get("hook_event").and_then(Value::as_str), Some("SessionStart"));
        assert_eq!(h1.get("output").and_then(Value::as_str), Some("## The PR review gate"));
        assert_eq!(h1.get("stdout").and_then(Value::as_str), Some("## The PR review gate"));
        assert_eq!(h1.get("exit_code").and_then(Value::as_i64), Some(0));
        assert_eq!(h1.get("outcome").and_then(Value::as_str), Some("success"));
        assert_eq!(h1.get("session_id").and_then(Value::as_str), Some("s1"));
        assert_eq!(h1.get("transcript").and_then(Value::as_bool), Some(true));

        let h2 = &msgs[2].message;
        assert_eq!(
            h2.get("output").and_then(Value::as_str),
            Some("line one\nline two"),
            "the content array joins its lines"
        );
        assert_eq!(h2.get("hook_event").and_then(Value::as_str), Some("Setup"));

        let h3 = &msgs[3].message;
        assert_eq!(h3.get("outcome").and_then(Value::as_str), Some("error"));
        assert_eq!(
            h3.get("output").and_then(Value::as_str),
            Some("Failed with non-blocking status code"),
            "the stderr fallback when the combined output and stdout are empty"
        );
        assert_eq!(h3.get("exit_code").and_then(Value::as_i64), Some(141));

        let h4 = &msgs[4].message;
        assert_eq!(h4.get("outcome").and_then(Value::as_str), Some("cancelled"));
        assert_eq!(h4.get("output").and_then(Value::as_str), Some(""));

        let h5 = &msgs[5].message;
        assert_eq!(
            h5.get("output").and_then(Value::as_str),
            Some("plugin lines"),
            "the stdout fallback for a row with no combined content"
        );

        // And the rows still decode as the frames a page serves: the scan's
        // shape is only a carrier, the wire shape is what crosses.
        let frames = messages_from_rows(parse_session_messages(jsonl.as_bytes()).messages, "s1");
        assert!(
            matches!(&frames[0], forge_primitives::Message::HookResponse { output, exit_code, .. }
                if output == "## The PR review gate" && *exit_code == 0),
            "the row decodes to the typed live response"
        );
        assert!(
            matches!(&frames[1], forge_primitives::Message::Assistant { .. }),
            "and the assistant row keeps its place between the hook rows"
        );
        assert!(
            frames.iter().any(|frame| matches!(frame,
                forge_primitives::Message::HookResponse { hook_id, outcome, .. }
                    if hook_id == "h4" && outcome == "cancelled")),
            "and a cancelled run decodes as the same typed response, not the generic system shape"
        );
        let crossed = serde_json::to_value(&frames[0]).expect("the frame crosses");
        for (field, want) in [
            ("hook_id", "h1"),
            ("output", "## The PR review gate"),
            ("outcome", "success"),
            ("transcript", "true"),
        ] {
            assert_eq!(
                crossed.get(field).map(|held| held.to_string().trim_matches('"').to_owned()),
                Some(want.to_owned()),
                "{field} survives the decode"
            );
        }
    }

    /// The count has to come out of the same streamed pass that builds
    /// the messages. A transcript that has compacted is usually a large
    /// one (a session compacts because it is huge), so a second read of
    /// the file to count them would land squarely on the resume path for
    /// exactly the sessions this number is about.
    ///
    /// **And the row rides the list, not only the count.** A resumed
    /// conversation draws its boundaries the way a live one does, which needs
    /// the frame the fold reads: the count alone leaves a page saying
    /// "3 compactions" over a conversation with no boundary anywhere in it.
    #[test]
    fn parse_session_messages_counts_compaction_boundaries() {
        let jsonl = r#"{"type":"user","message":{"role":"user","content":"one"},"uuid":"u1","session_id":"s1"}
{"type":"system","subtype":"compact_boundary","uuid":"cb1","session_id":"s1","compactMetadata":{"trigger":"auto","preTokens":1002459,"postTokens":31253}}
{"type":"assistant","message":{"role":"assistant","content":[]},"uuid":"as1","session_id":"s1"}
{"type":"system","subtype":"compact_boundary","uuid":"cb2","session_id":"s1","compactMetadata":{"trigger":"manual","preTokens":53903,"postTokens":9149}}
{"type":"system","subtype":"init","uuid":"x1","session_id":"s1"}
"#;
        let history = parse_session_messages(jsonl.as_bytes());
        assert_eq!(
            history.compaction_count, 2,
            "both boundaries counted, the other system row not"
        );
        assert_eq!(
            history.messages.len(),
            4,
            "both boundary rows ride the message list, and the unread system row does not"
        );

        // The disk row spells its metadata the CLI's way (`compactMetadata`,
        // `preTokens`) and the decoder keys on the wire's snake_case nesting,
        // so a row handed on un-normalised falls to the generic bucket and
        // draws bare where a live boundary draws its counts.
        let boundary = history
            .messages
            .iter()
            .find(|message| message.uuid == "cb1")
            .expect("the first boundary is carried");
        assert_eq!(boundary.session_id, "s1");
        assert_eq!(
            boundary.message.get("compact_metadata"),
            Some(&serde_json::json!({
                "trigger": "auto",
                "pre_tokens": 1_002_459,
                "post_tokens": 31_253,
            })),
            "normalised to the shape the wire sends",
        );
        assert_eq!(
            serde_json::from_value::<forge_primitives::Message>(boundary.message.clone())
                .expect("the kept row decodes"),
            forge_primitives::Message::CompactBoundary {
                trigger: "auto".to_owned(),
                pre_tokens: 1_002_459,
                post_tokens: 31_253,
                uuid: "cb1".to_owned(),
                session_id: "s1".to_owned(),
                metadata_extras: forge_primitives::messages::Extras::new(),
                extras: forge_primitives::messages::Extras::new(),
            },
            "and decodes typed, which is what the fold reads",
        );
    }

    #[test]
    fn parse_session_messages_counts_no_boundaries_in_a_never_compacted_session() {
        let jsonl = r#"{"type":"user","message":{"role":"user","content":"one"},"uuid":"u1","session_id":"s1"}
"#;
        assert_eq!(parse_session_messages(jsonl.as_bytes()).compaction_count, 0);
    }

    /// A turn's hooks are durable only as a `system/stop_hook_summary` row,
    /// and the scan dropped every `system` row that was not a compaction
    /// boundary, so a view opening the page fresh had nothing to draw the
    /// hook row from. The row is carried in the shape the wire sends, so the
    /// same decoder turns it into the frame the fold reads.
    #[test]
    fn parse_session_messages_keeps_a_stop_hook_summary() {
        let jsonl = r#"{"type":"user","message":{"role":"user","content":"one"},"uuid":"u1","session_id":"s1"}
{"type":"system","subtype":"stop_hook_summary","hookCount":2,"hookInfos":[{"command":"just fmt","durationMs":1400},{"command":"just check","durationMs":62000}],"hookErrors":[],"hasOutput":true,"level":"suggestion","preventedContinuation":false,"stopReason":"","toolUseID":"toolu_hook","timestamp":"2026-09-04T00:57:38.264Z","uuid":"h1","sessionId":"s1","cwd":"/proj"}
{"type":"system","subtype":"init","uuid":"x1","session_id":"s1"}
"#;
        let history = parse_session_messages(jsonl.as_bytes());

        assert_eq!(history.messages.len(), 2, "the hook row joins the turn, and nothing else does");
        assert!(
            matches!(history.messages[1].kind, SessionMessageKind::System),
            "kept as the system row it is",
        );
        assert_eq!(
            history.messages[1].uuid, "h1",
            "under its own id, which is what a view keys on"
        );
        assert_eq!(history.messages[1].timestamp.as_deref(), Some("2026-09-04T00:57:38.264Z"));

        let frame: forge_primitives::Message =
            serde_json::from_value(history.messages[1].message.clone())
                .expect("the kept row decodes as the frame the wire sends");
        let forge_primitives::Message::StopHookSummary { actions, hook_infos, .. } = frame else {
            panic!("a hook summary");
        };
        assert_eq!(actions, 2, "carrying the count the row reports");
        assert_eq!(hook_infos.len(), 2, "and one entry per hook behind it");
        assert_eq!(hook_infos[0].command, "just fmt", "each naming what it ran");
        assert_eq!(hook_infos[1].duration_ms, Some(62000), "and how long it took");
    }

    /// **The read serves the frames the live conversation draws.** A
    /// permission refusal, an API error, an informational line and a
    /// notification are rows the CLI persists, and the scan kept no system
    /// row but its two - so a resumed seat or a reopened page lost lines the
    /// live view had. A top-level type this build has no variant for arrives
    /// as itself, exactly as the live decoder surfaces it, while the
    /// transcript's own bookkeeping rows still stay out of the conversation.
    #[test]
    fn parse_session_messages_keeps_the_system_rows_the_live_view_draws() {
        let jsonl = r#"{"type":"system","subtype":"permission_denied","tool_name":"Bash","tool_use_id":"toolu_1","decision_reason_type":"other","decision_reason":"Contains simple_expansion","message":"Contains simple_expansion","uuid":"p1","sessionId":"s1","timestamp":"2026-10-26T10:00:00.000Z"}
{"type":"system","subtype":"api_error","level":"error","error":{"message":"502 upstream request failed","status":502},"retryInMs":610,"uuid":"e1","sessionId":"s1","timestamp":"2026-10-26T10:00:01.000Z"}
{"type":"system","subtype":"informational","content":"Goal cleared","level":"warning","uuid":"i1","sessionId":"s1","timestamp":"2026-10-26T10:00:02.000Z"}
{"type":"system","subtype":"notification","text":"a stop hook errored","priority":"immediate","uuid":"n1","sessionId":"s1","timestamp":"2026-10-26T10:00:03.000Z"}
{"type":"widget_state","widget":{"name":"counter"},"uuid":"w1","sessionId":"s1","timestamp":"2026-10-26T10:00:04.000Z"}
{"type":"worktree-state","worktree":{},"uuid":"ws1"}
{"type":"mode","mode":"default","uuid":"m1"}
{"type":"queue-operation","operation":"enqueue","timestamp":"2026-11-03T12:00:58.593Z","sessionId":"s1","content":"a prompt the reader cancelled"}
{"type":"system","subtype":"turn_duration","durationMs":10,"uuid":"t1","sessionId":"s1"}
"#;
        let history = parse_session_messages(jsonl.as_bytes());
        assert_eq!(
            history.messages.iter().map(|row| row.uuid.as_str()).collect::<Vec<_>>(),
            // `t1` is a subtype with no face of its own yet: the client draws
            // it provisionally, so the read keeps it too.
            ["p1", "e1", "i1", "n1", "w1", "t1"],
            "the live rows are kept, and no bookkeeping row is",
        );
        assert!(
            !history.messages.iter().any(|row| {
                row.message.get("type").and_then(serde_json::Value::as_str)
                    == Some("queue-operation")
            }),
            "an enqueue-only queue-operation row is bookkeeping, so a cold read never serves it",
        );

        let messages = messages_from_rows(history.messages, "s1");
        assert!(
            matches!(
                &messages[0],
                forge_primitives::Message::PermissionDenied { tool_use_id, session_id, .. }
                    if tool_use_id == "toolu_1" && session_id == "s1"
            ),
            "the refusal decodes typed, under the session the caller names: {:?}",
            messages[0],
        );
        assert!(
            matches!(
                &messages[1],
                forge_primitives::Message::System { subtype, .. } if subtype == "api_error"
            ),
            "the api error decodes as the generic system frame: {:?}",
            messages[1],
        );
        assert!(
            matches!(
                &messages[2],
                forge_primitives::Message::System { subtype, .. } if subtype == "informational"
            ),
            "and so does the informational line: {:?}",
            messages[2],
        );
        assert!(
            matches!(
                &messages[3],
                forge_primitives::Message::Notification { text, .. }
                    if text == "a stop hook errored"
            ),
            "the notification decodes typed: {:?}",
            messages[3],
        );
        assert!(
            matches!(
                &messages[4],
                forge_primitives::Message::Unknown { type_str, raw }
                    if type_str == "widget_state" && raw["widget"]["name"] == "counter"
            ),
            "an unknown top-level row arrives as itself, payload and all: {:?}",
            messages[4],
        );
        assert!(
            matches!(&messages[5], forge_primitives::Message::TurnDuration { ms: 10, .. }),
            "a system subtype with no face of its own still round-trips: {:?}",
            messages[5],
        );

        let frame = |line: &str| -> forge_primitives::Message {
            serde_json::from_value(serde_json::from_str(line).expect("json"))
                .expect("a decodable frame")
        };
        for served in [
            r#"{"type":"system","subtype":"permission_denied","tool_name":"Bash","tool_use_id":"t","decision_reason_type":"other","decision_reason":"r","message":"r","uuid":"p","session_id":"s"}"#,
            r#"{"type":"system","subtype":"api_error","error":{"message":"down"},"uuid":"e","session_id":"s"}"#,
            r#"{"type":"system","subtype":"informational","content":"c","uuid":"i","session_id":"s"}"#,
            r#"{"type":"system","subtype":"notification","text":"t","uuid":"n","session_id":"s"}"#,
            // A typed system subtype the read keeps while another surface
            // draws it: the numbering has to count its row as the scan does.
            r#"{"type":"system","subtype":"task_progress","task_id":"t","description":"mapping","usage":{"total_tokens":1,"tool_uses":1,"duration_ms":1},"uuid":"tp","session_id":"s"}"#,
        ] {
            assert!(
                has_a_transcript_row(&frame(served)),
                "a frame the read serves counts as having a row: {served}",
            );
        }
        assert!(
            !has_a_transcript_row(&frame(
                r#"{"type":"system","subtype":"init","uuid":"t","session_id":"s"}"#,
            )),
            "a system subtype the read drops still counts as no row",
        );
        assert!(
            !has_a_transcript_row(&frame(
                r#"{"type":"system","subtype":"thinking_tokens","estimated_tokens":100,"estimated_tokens_delta":10,"uuid":"k","session_id":"s"}"#,
            )),
            "the thinking counter is a report figure, not a row",
        );
        assert!(
            has_a_transcript_row(&frame(
                r#"{"type":"system","subtype":"turn_duration","durationMs":1,"uuid":"t","session_id":"s"}"#,
            )),
            "and a subtype the client draws provisionally has its row",
        );
    }

    /// **One spelling of the session in a read frame.** The CLI writes
    /// `sessionId` on a row and the wire sends `session_id`; a kept row that
    /// carried both named the session twice, which is what a reader expanding
    /// a provisional row saw raw (Ved, 2026-10-28). The scan drops the row's
    /// own spelling, so the read frame is shaped like the live one.
    #[test]
    fn a_kept_system_row_carries_one_session_spelling() {
        let jsonl = r#"{"type":"system","subtype":"api_error","error":{"message":"down"},"uuid":"e1","sessionId":"s1","timestamp":"2026-10-28T10:00:00.000Z"}
{"type":"system","subtype":"stop_hook_summary","hookCount":0,"hookInfos":[],"hookErrors":[],"hasOutput":false,"level":"suggestion","preventedContinuation":false,"stopReason":"","toolUseID":"toolu_h","uuid":"h1","sessionId":"s1","timestamp":"2026-10-28T10:00:01.000Z"}
"#;
        let frames = messages_from_rows(parse_session_messages(jsonl.as_bytes()).messages, "s1");
        assert_eq!(frames.len(), 2, "both kept system rows reach the frame");
        for (at, frame) in frames.iter().enumerate() {
            let wire = serde_json::to_value(frame).expect("a frame serialises").to_string();
            assert!(
                !wire.contains("\"sessionId\""),
                "frame {at} carries the CLI's spelling as well as the wire's: {wire}",
            );
            assert_eq!(
                wire.matches("\"session_id\"").count(),
                1,
                "frame {at} names the session exactly once: {wire}",
            );
        }
    }

    /// **One list, two consumers.** The scan and the client filter the same
    /// `system` subtypes: a subtype in one and not the other either loses a
    /// row at the transcript or draws one live that a reopened conversation
    /// cannot serve. The two live in different languages, so this test reads
    /// the client's literal and compares it with this crate's const.
    #[test]
    fn the_system_skip_list_matches_the_clients() {
        const CLIENT: &str = concat!(env!("CARGO_MANIFEST_DIR"), "/../../client/src/chat/units.ts");
        const MARKER: &str = "const SYSTEM_DRAWN_ELSEWHERE: ReadonlySet<string> = new Set([";
        let source = std::fs::read_to_string(CLIENT)
            .unwrap_or_else(|err| panic!("the client's fold is readable at {CLIENT}: {err}"));
        let start =
            source.find(MARKER).unwrap_or_else(|| panic!("the client still declares {MARKER:?}"));
        let block = &source[start + MARKER.len()..];
        let end = block
            .find("]);")
            .unwrap_or_else(|| panic!("the client's skip list closes after its entries"));
        let mut theirs: Vec<String> = block[..end]
            .split(',')
            .filter_map(|entry| {
                let entry = entry.trim();
                let quoted = entry.strip_prefix('\'')?.strip_suffix('\'')?;
                (!quoted.is_empty()).then(|| quoted.to_owned())
            })
            .collect();
        assert!(
            !theirs.is_empty(),
            "the client's skip list parsed to nothing, so this test compared nothing",
        );
        theirs.sort();
        let mut ours: Vec<String> =
            SYSTEM_ROWS_DRAWN_ELSEWHERE.iter().map(|name| (*name).to_owned()).collect();
        ours.sort();
        assert_eq!(
            ours, theirs,
            "the scan's SYSTEM_ROWS_DRAWN_ELSEWHERE and the client's SYSTEM_DRAWN_ELSEWHERE \
             have diverged; a subtype belongs in both or neither",
        );
    }

    /// **Every row a transcript read keeps reaches the wire stamped**, which
    /// is what a page's clock is read from: an open page anchors a turn on
    /// the first row that carries a stamp, and the row that opens a turn is
    /// the reader's own - so a kind that loses its stamp is a turn that draws
    /// with no age at all until an assistant frame lands.
    #[test]
    fn every_row_a_transcript_read_keeps_reaches_the_wire_stamped() {
        let jsonl = r#"{"type":"user","message":{"role":"user","content":[{"type":"text","text":"hi"}]},"uuid":"u1","sessionId":"s1","timestamp":"2026-10-15T05:18:05.217Z"}
{"type":"attachment","attachment":{"type":"queued_command","prompt":"queued words","commandMode":"prompt"},"uuid":"a1","session_id":"s1","timestamp":"2026-10-15T05:19:00.100Z"}
{"type":"queue-operation","operation":"remove","timestamp":"2026-10-15T05:19:03.000Z","sessionId":"s1","content":"queued words"}
{"type":"assistant","message":{"id":"m1","role":"assistant","model":"claude-opus-5","content":[{"type":"text","text":"working"}]},"uuid":"a1","sessionId":"s1","timestamp":"2026-10-15T05:19:05.544Z"}
{"type":"system","subtype":"stop_hook_summary","hookCount":1,"hookInfos":[],"hookErrors":[],"hasOutput":false,"level":"suggestion","preventedContinuation":false,"stopReason":"","toolUseID":"tu1","uuid":"h1","sessionId":"s1","timestamp":"2026-10-15T05:19:40.632Z"}
{"type":"system","subtype":"compact_boundary","uuid":"b1","sessionId":"s1","compactMetadata":{"trigger":"auto","preTokens":1018466,"postTokens":19526},"timestamp":"2026-10-15T05:20:11.004Z"}
"#;
        let history = parse_session_messages(jsonl.as_bytes());
        let messages = messages_from_rows(history.messages, "s1");

        let stamps: Vec<(String, Option<String>)> = messages
            .iter()
            .map(|message| {
                let value = serde_json::to_value(message).expect("encode");
                let kind = value["type"].as_str().unwrap_or("?").to_owned();
                (kind, value["timestamp"].as_str().map(str::to_owned))
            })
            .collect();
        assert_eq!(
            stamps,
            vec![
                ("user".to_owned(), Some("2026-10-15T05:18:05.217Z".to_owned())),
                ("user".to_owned(), Some("2026-10-15T05:19:03.000Z".to_owned())),
                ("assistant".to_owned(), Some("2026-10-15T05:19:05.544Z".to_owned())),
                ("system".to_owned(), Some("2026-10-15T05:19:40.632Z".to_owned())),
                ("system".to_owned(), Some("2026-10-15T05:20:11.004Z".to_owned())),
            ],
            "each kept kind must carry the row's own stamp, in the row's own order",
        );
    }

    #[test]
    fn parse_session_messages_skips_attachment_without_queued_command() {
        // Other attachment subtypes (image, document, etc.) are not
        // user-bubble-worthy - keep skipping them.
        let jsonl = r#"{"type":"attachment","attachment":{"type":"image","source":{}},"uuid":"a1","session_id":"s1"}
{"type":"assistant","message":{"role":"assistant","content":[]},"uuid":"as1","session_id":"s1"}
"#;
        let msgs = parse_session_messages(jsonl.as_bytes()).messages;
        assert_eq!(msgs.len(), 1, "non-queued_command attachment must be skipped");
        assert!(matches!(msgs[0].kind, SessionMessageKind::Assistant));
    }

    /// A transcript can carry a block of records twice: the second copies
    /// share the `uuid`, `type` and message content of the first and differ
    /// only in metadata fields, and they sit right after a later run's
    /// session-state rows. The whole-file read handed both copies on, so a
    /// resumed or freshly-opened conversation drew the block twice. The
    /// `uuid` is the key that absorbs it - unique per record and stable
    /// across the pair.
    #[test]
    fn parse_session_messages_reads_a_duplicated_block_once() {
        let jsonl = r#"{"type":"user","message":{"role":"user","content":"first"},"uuid":"u1","session_id":"s1","timestamp":"2026-09-04T02:23:48.850Z"}
{"type":"assistant","message":{"id":"msg_01","role":"assistant","content":[{"type":"text","text":"answer"}]},"uuid":"a1","session_id":"s1","timestamp":"2026-09-04T02:23:50.062Z"}
{"type":"last-prompt","uuid":"lp1"}
{"type":"tag","tag":"t"}
{"type":"user","message":{"role":"user","content":"first"},"uuid":"u1","slug":"structured-pondering-nygaard","session_id":"s1","timestamp":"2026-09-04T02:23:48.850Z"}
{"type":"assistant","message":{"id":"msg_01","role":"assistant","content":[{"type":"text","text":"answer"}]},"uuid":"a1","slug":"structured-pondering-nygaard","session_id":"s1","timestamp":"2026-09-04T02:23:50.062Z"}
"#;
        let history = parse_session_messages(jsonl.as_bytes());
        assert_eq!(
            history.messages.iter().map(|row| row.uuid.as_str()).collect::<Vec<_>>(),
            ["u1", "a1"],
            "the duplicated block reads once, in its original order, and the session-state rows \
             stay out",
        );
    }

    /// The dedupe keys on `uuid`, so a boundary row counted from a duplicated
    /// block is one compaction, not two: the second copy is the same cut.
    #[test]
    fn parse_session_messages_counts_a_duplicated_boundary_once() {
        let jsonl = r#"{"type":"user","message":{"role":"user","content":"one"},"uuid":"u1","session_id":"s1"}
{"type":"system","subtype":"compact_boundary","uuid":"cb1","session_id":"s1","compactMetadata":{"trigger":"auto","preTokens":1002459,"postTokens":31253}}
{"type":"last-prompt","uuid":"lp1"}
{"type":"system","subtype":"compact_boundary","uuid":"cb1","slug":"s","session_id":"s1","compactMetadata":{"trigger":"auto","preTokens":1002459,"postTokens":31253}}
"#;
        let history = parse_session_messages(jsonl.as_bytes());
        assert_eq!(history.compaction_count, 1, "one boundary, written twice");
        assert_eq!(history.messages.len(), 2, "and drawn once");
    }

    fn session_info_with_tag(session_id: &str, tag: Option<&str>) -> SDKSessionInfo {
        SDKSessionInfo {
            session_id: session_id.to_string(),
            summary: "test".to_string(),
            last_modified: 0,
            file_size: None,
            custom_title: None,
            first_prompt: None,
            git_branch: None,
            cwd: None,
            storage_key: String::new(),
            tag: tag.map(str::to_string),
            created_at: None,
        }
    }

    #[test]
    fn parse_session_info_filter_excludes_worker_tag_by_default() {
        let content = "{\"type\":\"user\",\"timestamp\":\"2026-04-22T00:00:00.000Z\",\"message\":{\"content\":\"hi\"}}\n{\"type\":\"tag\",\"tag\":\"forge:worker:reviewer\"}\n".to_string();
        let lite = LiteSessionFile {
            mtime: 0,
            size: content.len() as u64,
            head: content.clone(),
            tail: content,
            tag: Some("forge:worker:reviewer".to_string()),
        };
        let info = parse_session_info_from_lite("abc", &lite, None).expect("some");
        assert_eq!(info.tag.as_deref(), Some("forge:worker:reviewer"));
        // The session is still parseable - filtering happens at the
        // list_sessions caller layer, not here.
        assert!(should_exclude_worker_tag(&info));
    }

    #[test]
    fn should_exclude_worker_tag_recognises_prefix() {
        let info = session_info_with_tag("s1", Some("forge:worker:reviewer"));
        assert!(should_exclude_worker_tag(&info));
    }

    #[test]
    fn should_exclude_worker_tag_passes_lead_and_untagged() {
        let lead = session_info_with_tag("s1", Some("forge:lead"));
        let untagged = session_info_with_tag("s2", None);
        assert!(!should_exclude_worker_tag(&lead));
        assert!(!should_exclude_worker_tag(&untagged));
    }

    /// A one-row transcript, tagged when `tag` is given.
    fn transcript(tag: Option<&str>) -> String {
        use std::fmt::Write as _;
        let mut body = "{\"type\":\"user\",\"timestamp\":\"2026-04-22T00:00:00.000Z\",\
                        \"message\":{\"content\":\"hi\"}}\n"
            .to_owned();
        if let Some(tag) = tag {
            let _ = writeln!(body, "{{\"type\":\"tag\",\"tag\":\"{tag}\"}}");
        }
        body
    }

    /// The listing's ids, in the order the scan answered them.
    fn session_ids(entries: &[SDKSessionInfo]) -> Vec<String> {
        entries.iter().map(|info| info.session_id.clone()).collect()
    }

    /// The scan's directory for `cwd`, which the listing reads.
    fn dir_for(config: &Path, cwd: &str) -> PathBuf {
        projects_dir_for(config).join(project_key_for_directory(Some(cwd)))
    }

    /// An explicit mtime, so an ordering assertion does not rest on files
    /// written inside the same millisecond.
    fn set_age(path: &Path, seconds: u64) {
        let when = std::time::SystemTime::now() - std::time::Duration::from_secs(seconds.max(1));
        fs::File::options()
            .write(true)
            .open(path)
            .expect("open for its mtime")
            .set_modified(when)
            .expect("set its mtime");
    }

    /// A listing is the seat's own sessions. A worker's label is what it
    /// can resume, and the directory cannot answer that alone: a non-git
    /// worker shares the project root with its lead, so the two seats'
    /// transcripts sit in one directory. Catches a filter that hides
    /// every worker tag whatever the seat, one that keeps another label's
    /// sessions, and one that matches a label by prefix - `alpha` must not
    /// offer `alpha-2`'s sessions, which a resume would adopt as its own.
    #[tokio::test]
    async fn a_listing_keeps_the_seats_own_sessions_and_no_others() {
        let tmp = tempfile::tempdir().unwrap();
        let cwd = tmp.path().join("project");
        fs::create_dir_all(&cwd).unwrap();
        let cwd = cwd.to_string_lossy().into_owned();
        let dir = dir_for(tmp.path(), &cwd);
        fs::create_dir_all(&dir).unwrap();
        write_session_jsonl(&dir, "lead-own", &transcript(None));
        write_session_jsonl(&dir, "alpha-own", &transcript(Some("forge:worker:alpha")));
        write_session_jsonl(&dir, "alpha-2-own", &transcript(Some("forge:worker:alpha-2")));
        write_session_jsonl(&dir, "beta-own", &transcript(Some("forge:worker:beta")));

        let lead = list_sessions(tmp.path(), Some(&cwd), None, 0, Workers::Hidden, None).await;
        assert_eq!(
            session_ids(&lead),
            vec!["lead-own"],
            "a lead's listing is the project's own sessions, never its workers'"
        );

        let worker =
            list_sessions(tmp.path(), Some(&cwd), None, 0, Workers::Only("alpha".to_owned()), None)
                .await;
        assert_eq!(
            session_ids(&worker),
            vec!["alpha-own"],
            "a worker's listing is the sessions its own label ran, not its prefix kin's"
        );
    }

    /// **A worker's listing does not read what its cached tags already rule
    /// out** (#2567). The head-and-tail read of every transcript in the
    /// directory was the whole cost of a `/resume` list on a non-git worker,
    /// which shares the project root with its lead. The answer is the same
    /// rows either way, so the skip is read off its own log line.
    #[tokio::test]
    async fn a_workers_listing_skips_what_its_cached_tags_rule_out() {
        let logs = LogCapture::default();
        let subscriber = tracing_subscriber::fmt()
            .with_max_level(tracing::Level::DEBUG)
            .with_ansi(false)
            .with_writer(logs.clone())
            .finish();
        let _guard = tracing::subscriber::set_default(subscriber);

        let tmp = tempfile::tempdir().unwrap();
        let cwd = tmp.path().join("project");
        fs::create_dir_all(&cwd).unwrap();
        let cwd = cwd.to_string_lossy().into_owned();
        let dir = dir_for(tmp.path(), &cwd);
        fs::create_dir_all(&dir).unwrap();
        write_session_jsonl(&dir, "lead-own", &transcript(None));
        write_session_jsonl(&dir, "beta-own", &transcript(Some("forge:worker:beta")));
        write_session_jsonl(&dir, "alpha-own", &transcript(Some("forge:worker:alpha")));

        // A boot scan's cache: one full pass, then exactly what it learned.
        let first = std::sync::Arc::new(SessionTagCache::new(std::collections::HashMap::new()));
        let read =
            list_sessions(tmp.path(), Some(&cwd), None, 0, Workers::Hidden, Some(&first)).await;
        assert_eq!(read.len(), 1, "precondition: a lead's listing hides its workers");
        let prior: std::collections::HashMap<String, SessionTagScan> =
            first.updates().into_iter().collect();
        assert_eq!(prior.len(), 3, "and the pass still read every transcript's tag");
        let warm = std::sync::Arc::new(SessionTagCache::new(prior));

        let worker = list_sessions(
            tmp.path(),
            Some(&cwd),
            None,
            0,
            Workers::Only("alpha".to_owned()),
            Some(&warm),
        )
        .await;
        assert_eq!(
            session_ids(&worker),
            vec!["alpha-own"],
            "the listing is still the label's own sessions",
        );
        let logged = String::from_utf8(logs.0.lock().clone()).unwrap();
        assert!(
            logged.contains("session_scan_narrowed") && logged.contains("skipped=2"),
            "the two transcripts the cache ruled out are named as skipped: {logged}",
        );

        // **A transcript that grew is read, not skipped.** The bytes since the
        // last scan can carry a `/new` re-tag, so a cache entry behind its
        // file proves nothing - and here the new tag makes the file the
        // worker's own.
        let mut retagged = transcript(Some("forge:worker:beta"));
        retagged.push_str("{\"type\":\"tag\",\"tag\":\"forge:worker:alpha\"}\n");
        write_session_jsonl(&dir, "beta-own", &retagged);
        let again = list_sessions(
            tmp.path(),
            Some(&cwd),
            None,
            0,
            Workers::Only("alpha".to_owned()),
            Some(&warm),
        )
        .await;
        // The scan orders by millisecond mtime, so rows written inside one
        // millisecond have no promised order among themselves.
        let mut again_ids = session_ids(&again);
        again_ids.sort();
        assert_eq!(
            again_ids,
            vec!["alpha-own", "beta-own"],
            "a transcript that changed after its scan must be read, not skipped",
        );

        // **A path the cache has never seen is read, not skipped.** The cache
        // cannot vouch for a transcript created after its scan.
        write_session_jsonl(&dir, "gamma-own", &transcript(Some("forge:worker:alpha")));
        let fresh = list_sessions(
            tmp.path(),
            Some(&cwd),
            None,
            0,
            Workers::Only("alpha".to_owned()),
            Some(&warm),
        )
        .await;
        let fresh_ids = session_ids(&fresh);
        assert!(
            fresh_ids.iter().any(|id| id.as_str() == "gamma-own"),
            "a transcript the cache has not scanned must be read: {fresh_ids:?}",
        );
    }

    /// **The cap bounds the reads, not only the answer** (#2567). A label
    /// with more runs than a page serves must not pay a head-and-tail read
    /// per transcript: the newest are read first and the scan stops when the
    /// answer is full.
    #[tokio::test]
    async fn a_capped_listing_reads_only_the_rows_it_serves() {
        let logs = LogCapture::default();
        let subscriber = tracing_subscriber::fmt()
            .with_max_level(tracing::Level::DEBUG)
            .with_ansi(false)
            .with_writer(logs.clone())
            .finish();
        let _guard = tracing::subscriber::set_default(subscriber);

        let tmp = tempfile::tempdir().unwrap();
        let cwd = tmp.path().join("project");
        fs::create_dir_all(&cwd).unwrap();
        let cwd = cwd.to_string_lossy().into_owned();
        let dir = dir_for(tmp.path(), &cwd);
        fs::create_dir_all(&dir).unwrap();
        // **Created oldest-first, so directory order and mtime order
        // disagree.** run-7 is written first and run-0 last; a scan that
        // trusts readdir's order answers the oldest rows and the assertion
        // below fails.
        for at in (0..8).rev() {
            let path = write_session_jsonl(
                &dir,
                &format!("run-{at}"),
                &transcript(Some("forge:worker:alpha")),
            );
            // run-0 is the newest, run-7 the oldest.
            set_age(&path, at + 1);
        }

        let listed = list_sessions(
            tmp.path(),
            Some(&cwd),
            Some(2),
            0,
            Workers::Only("alpha".to_owned()),
            None,
        )
        .await;
        assert_eq!(session_ids(&listed), vec!["run-0", "run-1"], "the two newest rows");

        let logged = String::from_utf8(logs.0.lock().clone()).unwrap();
        assert!(
            logged.contains("session_scan_read") && logged.contains("read=2"),
            "the scan read every transcript for a two-row page: {logged}",
        );

        // The control: no cap still reads the whole directory.
        let all =
            list_sessions(tmp.path(), Some(&cwd), None, 0, Workers::Only("alpha".to_owned()), None)
                .await;
        assert_eq!(all.len(), 8, "an uncapped listing reads every transcript");
    }

    /// The worker filter lands BEFORE the cap, so a listing is never
    /// shortened by rows it would not have carried. Catches the cap-first
    /// order: with the newest transcript a worker's, a cap of one spends
    /// its single slot on that row and answers nothing.
    #[tokio::test]
    async fn the_cap_is_applied_to_the_rows_the_listing_carries() {
        let tmp = tempfile::tempdir().unwrap();
        let cwd = tmp.path().join("project");
        fs::create_dir_all(&cwd).unwrap();
        let cwd = cwd.to_string_lossy().into_owned();
        let dir = dir_for(tmp.path(), &cwd);
        fs::create_dir_all(&dir).unwrap();
        let held = write_session_jsonl(&dir, "lead-own", &transcript(None));
        // The worker's row is the newest, so a cap spent before the filter
        // lands on it and the listing comes back empty.
        let newest =
            write_session_jsonl(&dir, "alpha-own", &transcript(Some("forge:worker:alpha")));
        set_age(&held, 600);
        set_age(&newest, 60);

        let capped = list_sessions(tmp.path(), Some(&cwd), Some(1), 0, Workers::Hidden, None).await;
        assert_eq!(
            session_ids(&capped),
            vec!["lead-own"],
            "the cap must be spent on the rows the listing carries"
        );
    }

    // -----------------------------------------------------------------
    // Tag-scan regression tests.
    //
    // The tag row is written ONCE at session start and persists for the
    // life of the JSONL; PR #167's `/new` flow can re-write a tag
    // mid-file. The lite head/tail window misses both cases on large
    // transcripts. These tests exercise the full-file scan path via
    // `read_session_info`.
    // -----------------------------------------------------------------

    fn write_session_jsonl(dir: &Path, name: &str, body: &str) -> PathBuf {
        let path = dir.join(format!("{name}.jsonl"));
        fs::write(&path, body).unwrap();
        path
    }

    #[test]
    fn read_session_info_finds_tag_at_start_of_large_file() {
        // The bug shape: tag row is written at session start and sits
        // permanently near byte 0. Tail-only extraction scans only
        // the last LITE_READ_BUF_SIZE bytes, so for any transcript
        // > ~130 KB the tag falls outside the tail window and resume
        // sees `info.tag = None`. Build a transcript well past two
        // window-widths so the bug is unambiguous.
        let window = usize::try_from(LITE_READ_BUF_SIZE).unwrap();
        let tmp = tempfile::tempdir().unwrap();
        let mut body = String::new();
        body.push_str(
            "{\"type\":\"user\",\"timestamp\":\"2026-04-22T00:00:00.000Z\",\"message\":{\"content\":\"hello\"}}\n",
        );
        body.push_str("{\"type\":\"tag\",\"tag\":\"forge:worker:reviewer\"}\n");
        let filler = "{\"type\":\"assistant\",\"message\":{\"content\":\"x\"}}\n";
        while body.len() < window * 3 {
            body.push_str(filler);
        }
        let path = write_session_jsonl(tmp.path(), "abc", &body);

        let info = read_session_info(&path, None).expect("session info parsed");
        assert_eq!(info.tag.as_deref(), Some("forge:worker:reviewer"));
    }

    #[test]
    fn read_session_info_finds_tag_mid_file() {
        // The `/new` case (PR #167): user fires `/new` mid-session and
        // the CLI appends a fresh tag row at the current cursor. For a
        // large transcript that's neither head nor tail, it lands in
        // the middle, where neither window sees it.
        let window = usize::try_from(LITE_READ_BUF_SIZE).unwrap();
        let tmp = tempfile::tempdir().unwrap();
        let mut body = String::new();
        body.push_str(
            "{\"type\":\"user\",\"timestamp\":\"2026-04-22T00:00:00.000Z\",\"message\":{\"content\":\"hello\"}}\n",
        );
        let filler = "{\"type\":\"assistant\",\"message\":{\"content\":\"x\"}}\n";
        // Push past the head window with filler, then plant the tag.
        while body.len() < window + 4_096 {
            body.push_str(filler);
        }
        body.push_str("{\"type\":\"tag\",\"tag\":\"forge:worker:tester\"}\n");
        // Then push past the tail window with more filler so the tag
        // sits in the middle region neither head nor tail covers.
        while body.len() < window * 3 {
            body.push_str(filler);
        }
        let path = write_session_jsonl(tmp.path(), "abc", &body);

        let info = read_session_info(&path, None).expect("session info parsed");
        assert_eq!(info.tag.as_deref(), Some("forge:worker:tester"));
    }

    /// **The title read does not scan for the tag.** A tag scan covers the
    /// whole file, and the caller that reads one transcript for its title (a
    /// catalog row mirrored at Connected) must not pay for bytes it does not
    /// use; routing that call back through the full read would restore the
    /// scan, and the tag on the answer is what says so.
    #[test]
    fn a_title_read_carries_the_title_and_not_the_tag() {
        let tmp = tempfile::tempdir().unwrap();
        let tagged = "{\"type\":\"user\",\"message\":{\"content\":\"the work I left\"}}\n\
                      {\"type\":\"tag\",\"tag\":\"forge:worker:gone\"}\n";
        let path = write_session_jsonl(tmp.path(), "abc", tagged);

        let info = read_session_info_at(&path).expect("a readable transcript");

        assert_eq!(info.summary, "the work I left", "the title is read");
        assert_eq!(info.tag, None, "and the tag is not, because that scan is the whole file");
    }

    /// A transcript that shrank was truncated or replaced, so an offset
    /// recorded against the old bytes describes nothing in the new ones.
    /// Resuming past the end of the replacement would read no bytes at
    /// all and keep answering with a tag the file no longer contains.
    #[test]
    fn a_shrunk_transcript_is_scanned_again_rather_than_resumed() {
        let tmp = tempfile::tempdir().unwrap();
        let tagged = "{\"type\":\"user\",\"message\":{\"content\":\"hi\"}}\n\
                      {\"type\":\"tag\",\"tag\":\"forge:worker:gone\"}\n";
        let path = write_session_jsonl(tmp.path(), "abc", tagged);

        let cache = SessionTagCache::default();
        assert_eq!(
            read_session_lite(&path, Some(&cache), true).unwrap().tag.as_deref(),
            Some("forge:worker:gone"),
            "the first scan must find the tag it will later be asked to forget"
        );

        // Replay the cache the store would have handed back, then replace
        // the file with shorter, untagged content.
        let primed = SessionTagCache::new(cache.updates().into_iter().collect());
        std::fs::write(&path, "{\"type\":\"user\",\"message\":{\"content\":\"x\"}}\n").unwrap();

        assert_eq!(
            read_session_lite(&path, Some(&primed), true).unwrap().tag,
            None,
            "a replaced transcript must be re-scanned, not answered from the old offset"
        );
    }

    /// A resumed scan must agree with a whole-file one for every split
    /// point, and the hard case is a split inside a line: the writer had
    /// only flushed half a tag row when the previous scan ran. Crediting
    /// the file's length rather than the last newline would restart in
    /// the middle of that row, fail the prefix test, and lose the tag
    /// for good - it is never re-read.
    #[test]
    fn a_resumed_scan_sees_a_tag_that_was_half_written_when_it_stopped() {
        let settled = "{\"type\":\"user\",\"message\":{\"content\":\"hi\"}}\n";
        let tag_row = "{\"type\":\"tag\",\"tag\":\"forge-worker-x\"}\n";
        // The first scan sees a whole line plus the front of the tag row.
        let torn = &tag_row[..20];
        let (found, _project, state) =
            scan_tag_from(std::io::Cursor::new(format!("{settled}{torn}")), None);
        assert_eq!(found, None, "half a tag row carries no closing quote to parse");
        assert_eq!(
            state.scanned_len,
            settled.len() as u64,
            "an unterminated line must not be credited, or it is skipped on resume"
        );

        // The writer finishes the row; the next scan resumes from the
        // recorded offset, exactly as the production path seeks.
        let whole = format!("{settled}{tag_row}");
        let resume_at = usize::try_from(state.scanned_len).unwrap();
        let (resumed, _project, _) =
            scan_tag_from(std::io::Cursor::new(&whole[resume_at..]), Some(state));
        assert_eq!(
            resumed.as_deref(),
            Some("forge-worker-x"),
            "a resumed scan must find the completed row a whole-file scan would"
        );
        assert_eq!(
            resumed,
            scan_tag_from(std::io::Cursor::new(whole.as_str()), None).0,
            "resuming must agree with scanning the whole file"
        );
    }

    #[test]
    fn read_session_info_picks_last_tag_when_multiple() {
        // Preserve PR #167 semantics: when the JSONL carries multiple
        // tag rows (original spawn tag + later `/new` re-tag), the
        // most recent one wins. File must be larger than two window
        // widths so neither tag lands in head or tail, otherwise the
        // old tail-only path would coincidentally pick the right one
        // and mask a behavioural regression.
        let window = usize::try_from(LITE_READ_BUF_SIZE).unwrap();
        let tmp = tempfile::tempdir().unwrap();
        let mut body = String::new();
        body.push_str(
            "{\"type\":\"user\",\"timestamp\":\"2026-04-22T00:00:00.000Z\",\"message\":{\"content\":\"hi\"}}\n",
        );
        body.push_str("{\"type\":\"tag\",\"tag\":\"first\"}\n");
        let filler = "{\"type\":\"assistant\",\"message\":{\"content\":\"x\"}}\n";
        while body.len() < window + 4_096 {
            body.push_str(filler);
        }
        body.push_str("{\"type\":\"tag\",\"tag\":\"second\"}\n");
        while body.len() < window * 3 {
            body.push_str(filler);
        }
        let path = write_session_jsonl(tmp.path(), "abc", &body);

        let info = read_session_info(&path, None).expect("session info parsed");
        assert_eq!(info.tag.as_deref(), Some("second"));
    }

    #[test]
    fn read_session_info_reports_storage_dir_key() {
        // The storage key is the `projects/<KEY>/` dir the transcript
        // physically lives in - ground truth, unlike the head-read cwd.
        let tmp = tempfile::tempdir().unwrap();
        let storage_key = "-Users-me-Projects-playground--claude-worktrees-gpt-tutor";
        let project_dir = tmp.path().join(storage_key);
        fs::create_dir_all(&project_dir).unwrap();
        let path = write_session_jsonl(
            &project_dir,
            "abc",
            "{\"type\":\"user\",\"timestamp\":\"2026-04-22T00:00:00.000Z\",\"message\":{\"content\":\"hi\"}}\n",
        );

        let info = read_session_info(&path, None).expect("session info parsed");
        assert_eq!(
            info.storage_key, storage_key,
            "the scanned session reports the projects/<KEY>/ dir it lives in",
        );
    }

    /// A sub-agent's own frames live only in the CLI's sidecar
    /// (`<session>/subagents/agent-*.jsonl`), which nothing read: after a
    /// resume the dispatch's card held no calls (#1763). The transcript read
    /// folds each sidecar under the first dispatch its chain answers - where
    /// the dispatch ran, not at the end - which is the join the card fold
    /// reads.
    #[test]
    fn get_session_messages_folds_subagent_sidecars_under_their_dispatch() {
        let config_dir = tempfile::tempdir().unwrap();
        let session_id = "550e8400-e29b-41d4-a716-446655440000";
        let directory = "/Users/me/Projects/playground";
        let dir = project_dir_for(config_dir.path(), directory);
        fs::create_dir_all(&dir).unwrap();
        write_session_jsonl(
            &dir,
            session_id,
            "{\"type\":\"assistant\",\"message\":{\"role\":\"assistant\",\"content\":[{\"type\":\"tool_use\",\"id\":\"toolu_dispatch\",\"name\":\"Agent\",\"input\":{\"prompt\":\"explore\"}}]},\"uuid\":\"a1\"}\n\
             {\"type\":\"user\",\"message\":{\"role\":\"user\",\"content\":[{\"type\":\"tool_result\",\"tool_use_id\":\"toolu_dispatch\",\"content\":\"done\"}]},\"uuid\":\"u1\"}\n\
             {\"type\":\"assistant\",\"message\":{\"role\":\"assistant\",\"content\":[{\"type\":\"text\",\"text\":\"after\"}]},\"uuid\":\"a2\"}\n",
        );
        // The CLI's own layout: one sidecar per sub-agent, the dispatch it
        // ran under named in the sibling meta file.
        let sidecars = dir.join(session_id).join("subagents");
        fs::create_dir_all(&sidecars).unwrap();
        fs::write(
            sidecars.join("agent-abc.meta.json"),
            "{\"agentType\":\"Explore\",\"description\":\"map it\",\"toolUseId\":\"toolu_dispatch\"}",
        )
        .unwrap();
        fs::write(
            sidecars.join("agent-abc.jsonl"),
            "{\"type\":\"assistant\",\"isSidechain\":true,\"message\":{\"role\":\"assistant\",\"content\":[{\"type\":\"tool_use\",\"id\":\"toolu_child\",\"name\":\"Bash\",\"input\":{\"command\":\"ls\"}}]},\"uuid\":\"c1\"}\n\
             {\"type\":\"user\",\"isSidechain\":true,\"message\":{\"role\":\"user\",\"content\":[{\"type\":\"tool_result\",\"tool_use_id\":\"toolu_child\",\"content\":\"out\"}]},\"uuid\":\"c2\"}\n",
        )
        .unwrap();
        // Two sidecars can share one dispatch, and both join under it.
        fs::write(
            sidecars.join("agent-def.meta.json"),
            "{\"agentType\":\"Explore\",\"description\":\"more\",\"toolUseId\":\"toolu_dispatch\"}",
        )
        .unwrap();
        fs::write(
            sidecars.join("agent-def.jsonl"),
            "{\"type\":\"assistant\",\"isSidechain\":true,\"message\":{\"role\":\"assistant\",\"content\":[{\"type\":\"text\",\"text\":\"second\"}]},\"uuid\":\"d1\"}\n",
        )
        .unwrap();
        // A sidecar whose dispatch is not in the transcript joins nothing;
        // a meta with no sidecar file is skipped rather than failing the read.
        fs::write(
            sidecars.join("agent-zzz.meta.json"),
            "{\"agentType\":\"Explore\",\"toolUseId\":\"toolu_absent\"}",
        )
        .unwrap();
        fs::write(
            sidecars.join("agent-zzz.jsonl"),
            "{\"type\":\"assistant\",\"isSidechain\":true,\"message\":{\"role\":\"assistant\",\"content\":[{\"type\":\"text\",\"text\":\"orphan\"}]},\"uuid\":\"x1\"}\n",
        )
        .unwrap();
        fs::write(
            sidecars.join("agent-nofile.meta.json"),
            "{\"agentType\":\"Explore\",\"toolUseId\":\"toolu_dispatch\"}",
        )
        .unwrap();
        // A sidecar with no meta at all, a meta that does not parse, and a
        // meta whose dispatch id is empty are each dropped rather than
        // failing the read or joining anything.
        for (stem, frame) in [("agent-lone", "y1"), ("agent-bad", "y2"), ("agent-empty", "y3")] {
            if stem != "agent-lone" {
                let meta = if stem == "agent-bad" {
                    "not json".to_owned()
                } else {
                    "{\"agentType\":\"Explore\",\"toolUseId\":\"\"}".to_owned()
                };
                fs::write(sidecars.join(format!("{stem}.meta.json")), meta).unwrap();
            }
            fs::write(
                sidecars.join(format!("{stem}.jsonl")),
                format!(
                    "{{\"type\":\"assistant\",\"isSidechain\":true,\"message\":{{\"role\":\"assistant\",\"content\":[{{\"type\":\"text\",\"text\":\"{frame}\"}}]}},\"uuid\":\"{frame}\"}}\n"
                ),
            )
            .unwrap();
        }

        let messages =
            get_session_messages(config_dir.path(), session_id, Some(directory)).messages;

        let uuids: Vec<_> = messages.iter().map(|message| message.uuid.as_str()).collect();
        assert_eq!(uuids.len(), 6, "the transcript's three rows plus three sidecar frames");
        assert_eq!(&uuids[..2], &["a1", "u1"], "the dispatch and its result come first");
        assert_eq!(uuids[5], "a2", "and the conversation's later row stays last");
        let mut folded = uuids[2..5].to_vec();
        folded.sort_unstable();
        assert_eq!(
            folded,
            vec!["c1", "c2", "d1"],
            "every sidecar sharing the dispatch sits where it ran, not at the end",
        );
        for absent in ["x1", "y1", "y2", "y3"] {
            assert!(
                !uuids.contains(&absent),
                "'{absent}' has no joinable dispatch and must not be folded",
            );
        }
        assert_eq!(
            messages.iter().filter(|message| message.parent_tool_use_id.is_none()).count(),
            3,
            "the transcript's own rows stay the main thread's",
        );
        for message in messages.iter().filter(|message| message.parent_tool_use_id.is_some()) {
            assert_eq!(
                message.parent_tool_use_id.as_deref(),
                Some("toolu_dispatch"),
                "each folded frame names the dispatch id itself, not the stem or the agent id",
            );
        }
        // The chat-row read takes the session's own rows only: a folded
        // sidecar frame here would be drawn as a chat row no card owns.
        let main_thread =
            read_session_messages_main_thread(config_dir.path(), session_id, Some(directory))
                .messages;
        let main_uuids: Vec<_> = main_thread.iter().map(|message| message.uuid.as_str()).collect();
        assert_eq!(
            main_uuids,
            vec!["a1", "u1", "a2"],
            "the main-thread read leaves every sub-agent frame out",
        );
    }

    /// **A nested dispatch's sidecar folds under the dispatch that owns it.**
    /// A nested dispatch's `Task` call is a call inside the outer agent's
    /// sidecar, never a row the transcript carries, so its sidecar used to
    /// join against whatever the growing message list held: the nested id
    /// when the parent had been spliced first, nothing when the child was
    /// read first. The fold now walks the dispatch chain to the first
    /// dispatch the transcript answers, three levels down included, and
    /// every nested row lands under that expansion.
    ///
    /// The read is taken twice with the meta mtimes swapped: the child may
    /// sort before or after its parent, and neither order may move a row.
    /// Each block is emitted inside the row that fired it, so the outer
    /// agent's first frame, then the nested agent's work, then the outer
    /// agent's result read in the order they ran.
    #[test]
    fn get_session_messages_folds_a_nested_sidecar_under_its_owner() {
        let config_dir = tempfile::tempdir().unwrap();
        let session_id = "550e8400-e29b-41d4-a716-446655440002";
        let directory = "/Users/me/Projects/playground";
        let dir = project_dir_for(config_dir.path(), directory);
        fs::create_dir_all(&dir).unwrap();
        write_session_jsonl(
            &dir,
            session_id,
            "{\"type\":\"assistant\",\"message\":{\"role\":\"assistant\",\"content\":[{\"type\":\"tool_use\",\"id\":\"toolu_dispatch\",\"name\":\"Agent\",\"input\":{\"prompt\":\"explore\"}}]},\"uuid\":\"a1\"}\n\
             {\"type\":\"user\",\"message\":{\"role\":\"user\",\"content\":[{\"type\":\"tool_result\",\"tool_use_id\":\"toolu_dispatch\",\"content\":\"done\"}]},\"uuid\":\"u1\"}\n\
             {\"type\":\"assistant\",\"message\":{\"role\":\"assistant\",\"content\":[{\"type\":\"text\",\"text\":\"after\"}]},\"uuid\":\"a2\"}\n",
        );
        let sidecars = dir.join(session_id).join("subagents");
        fs::create_dir_all(&sidecars).unwrap();
        // The outer agent's sidecar: it fired the nested dispatch and holds
        // the result that answers it, which is why the nested sidecar used
        // to join at all.
        fs::write(
            sidecars.join("agent-outer.meta.json"),
            "{\"agentType\":\"Explore\",\"toolUseId\":\"toolu_dispatch\"}",
        )
        .unwrap();
        fs::write(
            sidecars.join("agent-outer.jsonl"),
            "{\"type\":\"assistant\",\"isSidechain\":true,\"message\":{\"role\":\"assistant\",\"content\":[{\"type\":\"tool_use\",\"id\":\"toolu_inner\",\"name\":\"Agent\",\"input\":{\"prompt\":\"inner\"}}]},\"uuid\":\"o1\"}\n\
             {\"type\":\"user\",\"isSidechain\":true,\"message\":{\"role\":\"user\",\"content\":[{\"type\":\"tool_result\",\"tool_use_id\":\"toolu_inner\",\"content\":\"inner done\"}]},\"uuid\":\"o2\"}\n",
        )
        .unwrap();
        // The nested agent's sidecar, newer than its parent's.
        fs::write(
            sidecars.join("agent-nested.meta.json"),
            "{\"agentType\":\"Explore\",\"toolUseId\":\"toolu_inner\",\"parentAgentId\":\"outer\",\"spawnDepth\":2}",
        )
        .unwrap();
        fs::write(
            sidecars.join("agent-nested.jsonl"),
            "{\"type\":\"assistant\",\"isSidechain\":true,\"message\":{\"role\":\"assistant\",\"content\":[{\"type\":\"tool_use\",\"id\":\"toolu_deep\",\"name\":\"Bash\",\"input\":{\"command\":\"ls\"}}]},\"uuid\":\"n1\"}\n\
             {\"type\":\"user\",\"isSidechain\":true,\"message\":{\"role\":\"user\",\"content\":[{\"type\":\"tool_result\",\"tool_use_id\":\"toolu_deep\",\"content\":\"out\"}]},\"uuid\":\"n2\"}\n",
        )
        .unwrap();
        // The third level: its dispatch is a call inside the nested agent's
        // sidecar, so only the chain reaches the card.
        fs::write(
            sidecars.join("agent-deep.meta.json"),
            "{\"agentType\":\"Explore\",\"toolUseId\":\"toolu_deep\",\"parentAgentId\":\"nested\",\"spawnDepth\":3}",
        )
        .unwrap();
        fs::write(
            sidecars.join("agent-deep.jsonl"),
            "{\"type\":\"assistant\",\"isSidechain\":true,\"message\":{\"role\":\"assistant\",\"content\":[{\"type\":\"text\",\"text\":\"deep\"}]},\"uuid\":\"d1\"}\n",
        )
        .unwrap();

        let read = || get_session_messages(config_dir.path(), session_id, Some(directory)).messages;
        let assert_folded = |messages: &[SessionMessage]| {
            let uuids: Vec<_> = messages.iter().map(|message| message.uuid.as_str()).collect();
            assert_eq!(
                uuids,
                vec!["a1", "u1", "o1", "n1", "d1", "n2", "o2", "a2"],
                "the transcript's rows with each block emitted inside the row that fired it",
            );
            for message in messages.iter().filter(|message| message.parent_tool_use_id.is_some()) {
                assert_eq!(
                    message.parent_tool_use_id.as_deref(),
                    Some("toolu_dispatch"),
                    "every nested frame names the card that owns the dispatch, not the cardless one",
                );
            }
        };

        // Children newer than their parents: the shape the old growing-list
        // fold skipped.
        set_age(&sidecars.join("agent-outer.meta.json"), 600);
        set_age(&sidecars.join("agent-nested.meta.json"), 60);
        set_age(&sidecars.join("agent-deep.meta.json"), 30);
        assert_folded(&read());

        // And the reverse, the measured anomaly (6 of 162 nested sidecars) where
        // the old fold joined the child against its parent's spliced rows and
        // stamped the nested id.
        set_age(&sidecars.join("agent-outer.meta.json"), 30);
        set_age(&sidecars.join("agent-nested.meta.json"), 60);
        set_age(&sidecars.join("agent-deep.meta.json"), 600);
        assert_folded(&read());
    }

    /// **A nested sidecar folds even when its parent's body fell past the
    /// budget.** The chain comes off the metas, not the bodies: reading only
    /// the sidecars inside the budget used to strand a child whose parent was
    /// skipped, and its work vanished from the resume with no row.
    #[test]
    fn a_nested_sidecar_folds_when_its_parent_body_was_skipped() {
        let config_dir = tempfile::tempdir().unwrap();
        let session_id = "550e8400-e29b-41d4-a716-446655440004";
        let directory = "/Users/me/Projects/playground";
        let dir = project_dir_for(config_dir.path(), directory);
        fs::create_dir_all(&dir).unwrap();
        write_session_jsonl(
            &dir,
            session_id,
            "{\"type\":\"assistant\",\"message\":{\"role\":\"assistant\",\"content\":[{\"type\":\"tool_use\",\"id\":\"toolu_dispatch\",\"name\":\"Agent\",\"input\":{\"prompt\":\"explore\"}}]},\"uuid\":\"a1\"}\n\
             {\"type\":\"user\",\"message\":{\"role\":\"user\",\"content\":[{\"type\":\"tool_result\",\"tool_use_id\":\"toolu_dispatch\",\"content\":\"done\"}]},\"uuid\":\"u1\"}\n",
        );
        let sidecars = dir.join(session_id).join("subagents");
        fs::create_dir_all(&sidecars).unwrap();
        fs::write(
            sidecars.join("agent-outer.meta.json"),
            "{\"agentType\":\"Explore\",\"toolUseId\":\"toolu_dispatch\"}",
        )
        .unwrap();
        fs::write(
            sidecars.join("agent-outer.jsonl"),
            format!(
                "{{\"type\":\"assistant\",\"isSidechain\":true,\"message\":{{\"role\":\"assistant\",\"content\":[{{\"type\":\"tool_use\",\"id\":\"toolu_inner\",\"name\":\"Agent\",\"input\":{{\"prompt\":\"inner\"}}}}]}},\"uuid\":\"o1\"}}\n{}",
                " ".repeat(400)
            ),
        )
        .unwrap();
        fs::write(
            sidecars.join("agent-nested.meta.json"),
            "{\"agentType\":\"Explore\",\"toolUseId\":\"toolu_inner\",\"parentAgentId\":\"outer\",\"spawnDepth\":2}",
        )
        .unwrap();
        fs::write(
            sidecars.join("agent-nested.jsonl"),
            "{\"type\":\"assistant\",\"isSidechain\":true,\"message\":{\"role\":\"assistant\",\"content\":[{\"type\":\"text\",\"text\":\"kept\"}]},\"uuid\":\"n1\"}\n",
        )
        .unwrap();
        // A second child of the same skipped parent: both become roots, and
        // their order must be the one they fired in (oldest first).
        fs::write(
            sidecars.join("agent-second.meta.json"),
            "{\"agentType\":\"Explore\",\"toolUseId\":\"toolu_other\",\"parentAgentId\":\"outer\",\"spawnDepth\":2}",
        )
        .unwrap();
        fs::write(
            sidecars.join("agent-second.jsonl"),
            "{\"type\":\"assistant\",\"isSidechain\":true,\"message\":{\"role\":\"assistant\",\"content\":[{\"type\":\"text\",\"text\":\"older\"}]},\"uuid\":\"m1\"}\n",
        )
        .unwrap();
        set_age(&sidecars.join("agent-outer.meta.json"), 600);
        set_age(&sidecars.join("agent-nested.meta.json"), 60);
        set_age(&sidecars.join("agent-second.meta.json"), 120);

        // The budget fits the children's bodies only: the parent's is skipped.
        let transcript = dir.join(format!("{session_id}.jsonl"));
        let (sidecars, links) = subagent_sidecars_with_budget(session_id, &transcript, 300);
        assert!(
            sidecars.iter().all(|(dispatch, _)| dispatch != "toolu_dispatch"),
            "the parent's body is past the budget",
        );
        assert_eq!(links.len(), 3, "but every meta is read for the chain");

        let mut messages = parse_session_messages(fs::File::open(&transcript).unwrap()).messages;
        fold_subagent_sidecars(&mut messages, sidecars, &links, session_id);

        assert_eq!(
            messages.iter().map(|message| message.uuid.as_str()).collect::<Vec<_>>(),
            vec!["a1", "u1", "m1", "n1"],
            "both children fold, the older root first, although the parent's body was skipped",
        );
        for message in messages.iter().filter(|message| message.parent_tool_use_id.is_some()) {
            assert_eq!(
                message.parent_tool_use_id.as_deref(),
                Some("toolu_dispatch"),
                "every child is stamped with the dispatch the transcript answers",
            );
        }
    }

    /// **A child meta with no `parentAgentId` still resolves through the
    /// parent body.** The parent's own rows name the child's dispatch, and
    /// that link is enough.
    #[test]
    fn a_child_meta_without_a_parent_agent_id_resolves_through_the_body() {
        let config_dir = tempfile::tempdir().unwrap();
        let session_id = "550e8400-e29b-41d4-a716-446655440006";
        let directory = "/Users/me/Projects/playground";
        let dir = project_dir_for(config_dir.path(), directory);
        fs::create_dir_all(&dir).unwrap();
        write_session_jsonl(
            &dir,
            session_id,
            "{\"type\":\"assistant\",\"message\":{\"role\":\"assistant\",\"content\":[{\"type\":\"tool_use\",\"id\":\"toolu_dispatch\",\"name\":\"Agent\",\"input\":{\"prompt\":\"explore\"}}]},\"uuid\":\"a1\"}\n\
             {\"type\":\"user\",\"message\":{\"role\":\"user\",\"content\":[{\"type\":\"tool_result\",\"tool_use_id\":\"toolu_dispatch\",\"content\":\"done\"}]},\"uuid\":\"u1\"}\n",
        );
        let sidecars = dir.join(session_id).join("subagents");
        fs::create_dir_all(&sidecars).unwrap();
        fs::write(
            sidecars.join("agent-outer.meta.json"),
            "{\"agentType\":\"Explore\",\"toolUseId\":\"toolu_dispatch\"}",
        )
        .unwrap();
        fs::write(
            sidecars.join("agent-outer.jsonl"),
            "{\"type\":\"assistant\",\"isSidechain\":true,\"message\":{\"role\":\"assistant\",\"content\":[{\"type\":\"tool_use\",\"id\":\"toolu_child\",\"name\":\"Agent\",\"input\":{\"prompt\":\"inner\"}}]},\"uuid\":\"o1\"}\n\
             {\"type\":\"user\",\"isSidechain\":true,\"message\":{\"role\":\"user\",\"content\":[{\"type\":\"tool_result\",\"tool_use_id\":\"toolu_child\",\"content\":\"ok\"}]},\"uuid\":\"o2\"}\n",
        )
        .unwrap();
        // No parentAgentId: only the parent body names this dispatch.
        fs::write(
            sidecars.join("agent-child.meta.json"),
            "{\"agentType\":\"Explore\",\"toolUseId\":\"toolu_child\"}",
        )
        .unwrap();
        fs::write(
            sidecars.join("agent-child.jsonl"),
            "{\"type\":\"assistant\",\"isSidechain\":true,\"message\":{\"role\":\"assistant\",\"content\":[{\"type\":\"text\",\"text\":\"child\"}]},\"uuid\":\"c1\"}\n",
        )
        .unwrap();
        set_age(&sidecars.join("agent-outer.meta.json"), 600);
        set_age(&sidecars.join("agent-child.meta.json"), 60);

        let messages =
            get_session_messages(config_dir.path(), session_id, Some(directory)).messages;

        assert_eq!(
            messages.iter().map(|message| message.uuid.as_str()).collect::<Vec<_>>(),
            vec!["a1", "u1", "o1", "c1", "o2"],
            "the child folds inside the row that fired it",
        );
        for message in messages.iter().filter(|message| message.parent_tool_use_id.is_some()) {
            assert_eq!(message.parent_tool_use_id.as_deref(), Some("toolu_dispatch"));
        }
    }

    /// **A chain that loops, or loses its parent meta, joins nothing.** Both
    /// shapes are read off disk and either could be torn by a sync; the walk
    /// must terminate and the rows must not land under an id nothing draws.
    #[test]
    fn a_sidecar_chain_with_a_cycle_or_a_missing_parent_joins_nothing() {
        let config_dir = tempfile::tempdir().unwrap();
        let session_id = "550e8400-e29b-41d4-a716-446655440005";
        let directory = "/Users/me/Projects/playground";
        let dir = project_dir_for(config_dir.path(), directory);
        fs::create_dir_all(&dir).unwrap();
        write_session_jsonl(
            &dir,
            session_id,
            "{\"type\":\"assistant\",\"message\":{\"role\":\"assistant\",\"content\":[{\"type\":\"text\",\"text\":\"before\"}]},\"uuid\":\"a1\"}\n",
        );
        let sidecars = dir.join(session_id).join("subagents");
        fs::create_dir_all(&sidecars).unwrap();
        let write = |stem: &str, meta: &str| {
            fs::write(sidecars.join(format!("agent-{stem}.meta.json")), meta).unwrap();
            fs::write(
                sidecars.join(format!("agent-{stem}.jsonl")),
                format!(
                    "{{\"type\":\"assistant\",\"isSidechain\":true,\"message\":{{\"role\":\"assistant\",\"content\":[{{\"type\":\"text\",\"text\":\"{stem}\"}}]}},\"uuid\":\"{stem}\"}}\n"
                ),
            )
            .unwrap();
        };
        // Two sidecars that name each other: neither chain reaches an answer.
        write("a", "{\"agentType\":\"Explore\",\"toolUseId\":\"toolu_a\",\"parentAgentId\":\"b\"}");
        write("b", "{\"agentType\":\"Explore\",\"toolUseId\":\"toolu_b\",\"parentAgentId\":\"a\"}");
        // A child whose parent meta is absent.
        write(
            "c",
            "{\"agentType\":\"Explore\",\"toolUseId\":\"toolu_c\",\"parentAgentId\":\"ghost\"}",
        );

        let messages =
            get_session_messages(config_dir.path(), session_id, Some(directory)).messages;

        assert_eq!(
            messages.iter().map(|message| message.uuid.as_str()).collect::<Vec<_>>(),
            vec!["a1"],
            "a cycle and a missing parent both resolve to nothing, and nothing loops",
        );
        // The rows above would also vanish for a walk that answered a
        // non-answered id, so pin the resolution itself.
        let mut made_under = HashMap::new();
        made_under.insert("toolu_a".to_owned(), "toolu_b".to_owned());
        made_under.insert("toolu_b".to_owned(), "toolu_a".to_owned());
        made_under.insert("toolu_c".to_owned(), "toolu_ghost".to_owned());
        let mut answered = HashSet::new();
        answered.insert("toolu_top".to_owned());
        assert_eq!(
            owning_dispatch("toolu_a", &answered, &made_under),
            None,
            "a cycle resolves to nothing"
        );
        assert_eq!(owning_dispatch("toolu_b", &answered, &made_under), None);
        assert_eq!(
            owning_dispatch("toolu_c", &answered, &made_under),
            None,
            "a missing parent resolves to nothing",
        );
        made_under.insert("toolu_top_child".to_owned(), "toolu_top".to_owned());
        assert_eq!(
            owning_dispatch("toolu_top_child", &answered, &made_under),
            Some("toolu_top"),
            "a chain that reaches an answered dispatch resolves to it",
        );
    }

    /// A sidecar whose chain never reaches a dispatch the transcript answers
    /// joins nothing: folding it under an id no card owns would put rows in
    /// the conversation that draw nowhere.
    #[test]
    fn a_sidecar_nested_under_an_unanswered_dispatch_joins_nothing() {
        let config_dir = tempfile::tempdir().unwrap();
        let session_id = "550e8400-e29b-41d4-a716-446655440003";
        let directory = "/Users/me/Projects/playground";
        let dir = project_dir_for(config_dir.path(), directory);
        fs::create_dir_all(&dir).unwrap();
        // The outer dispatch has no result row: nothing answers it.
        write_session_jsonl(
            &dir,
            session_id,
            "{\"type\":\"assistant\",\"message\":{\"role\":\"assistant\",\"content\":[{\"type\":\"tool_use\",\"id\":\"toolu_dispatch\",\"name\":\"Agent\",\"input\":{\"prompt\":\"explore\"}}]},\"uuid\":\"a1\"}\n",
        );
        let sidecars = dir.join(session_id).join("subagents");
        fs::create_dir_all(&sidecars).unwrap();
        fs::write(
            sidecars.join("agent-outer.meta.json"),
            "{\"agentType\":\"Explore\",\"toolUseId\":\"toolu_dispatch\"}",
        )
        .unwrap();
        fs::write(
            sidecars.join("agent-outer.jsonl"),
            "{\"type\":\"assistant\",\"isSidechain\":true,\"message\":{\"role\":\"assistant\",\"content\":[{\"type\":\"tool_use\",\"id\":\"toolu_inner\",\"name\":\"Agent\",\"input\":{\"prompt\":\"inner\"}}]},\"uuid\":\"o1\"}\n",
        )
        .unwrap();
        fs::write(
            sidecars.join("agent-nested.meta.json"),
            "{\"agentType\":\"Explore\",\"toolUseId\":\"toolu_inner\"}",
        )
        .unwrap();
        fs::write(
            sidecars.join("agent-nested.jsonl"),
            "{\"type\":\"assistant\",\"isSidechain\":true,\"message\":{\"role\":\"assistant\",\"content\":[{\"type\":\"text\",\"text\":\"nested\"}]},\"uuid\":\"n1\"}\n",
        )
        .unwrap();

        let messages =
            get_session_messages(config_dir.path(), session_id, Some(directory)).messages;

        assert_eq!(
            messages.iter().map(|message| message.uuid.as_str()).collect::<Vec<_>>(),
            vec!["a1"],
            "neither sidecar has a row the transcript answers",
        );
    }

    /// The read budget bounds the parse: a sidecar larger than the budget is
    /// skipped while a smaller one is still read. A resume otherwise opens
    /// and parses every sidecar the session ever wrote, and most are older
    /// than the window the conversation keeps (#1763).
    #[test]
    fn subagent_sidecars_stop_at_the_read_budget() {
        let config_dir = tempfile::tempdir().unwrap();
        let session_id = "550e8400-e29b-41d4-a716-446655440001";
        let directory = "/Users/me/Projects/playground";
        let dir = project_dir_for(config_dir.path(), directory);
        fs::create_dir_all(&dir).unwrap();
        write_session_jsonl(&dir, session_id, "");
        let transcript = dir.join(format!("{session_id}.jsonl"));
        let sidecars = dir.join(session_id).join("subagents");
        fs::create_dir_all(&sidecars).unwrap();
        let write_sidecar = |stem: &str, uuid: &str, pad: usize| {
            fs::write(
                sidecars.join(format!("{stem}.meta.json")),
                "{\"agentType\":\"Explore\",\"toolUseId\":\"toolu_a\"}",
            )
            .unwrap();
            fs::write(
                sidecars.join(format!("{stem}.jsonl")),
                format!(
                    "{{\"type\":\"assistant\",\"isSidechain\":true,\"message\":{{\"role\":\"assistant\",\"content\":[{{\"type\":\"text\",\"text\":\"x\"}}]}},\"uuid\":\"{uuid}\"}}\n{}",
                    " ".repeat(pad)
                ),
            )
            .unwrap();
        };
        write_sidecar("agent-small", "s1", 0);
        write_sidecar("agent-large", "l1", 400);

        let (sidecars, _) = subagent_sidecars_with_budget(session_id, &transcript, 200);
        let uuids: Vec<_> =
            sidecars.iter().flat_map(|(_, rows)| rows.iter().map(|row| row.uuid.clone())).collect();

        assert!(uuids.iter().any(|uuid| uuid == "s1"), "a sidecar inside the budget is read");
        assert!(
            !uuids.iter().any(|uuid| uuid == "l1"),
            "a sidecar past the budget is skipped rather than parsed",
        );
    }

    /// **The budget spends on the newest sidecars first.** A sidecar past
    /// the budget keeps its dispatch alone, and the sidecars beyond the
    /// conversation window are the oldest ones - so a read that kept the
    /// oldest would bring back exactly the #1763 symptom at the newest edge.
    #[test]
    fn subagent_sidecars_spend_the_budget_on_the_newest_first() {
        let config_dir = tempfile::tempdir().unwrap();
        let session_id = "550e8400-e29b-41d4-a716-446655440002";
        let directory = "/Users/me/Projects/playground";
        let dir = project_dir_for(config_dir.path(), directory);
        fs::create_dir_all(&dir).unwrap();
        write_session_jsonl(&dir, session_id, "");
        let transcript = dir.join(format!("{session_id}.jsonl"));
        let sidecars = dir.join(session_id).join("subagents");
        fs::create_dir_all(&sidecars).unwrap();
        let write_sidecar = |stem: &str, uuid: &str| {
            fs::write(
                sidecars.join(format!("{stem}.meta.json")),
                "{\"agentType\":\"Explore\",\"toolUseId\":\"toolu_a\"}",
            )
            .unwrap();
            fs::write(
                sidecars.join(format!("{stem}.jsonl")),
                format!(
                    "{{\"type\":\"assistant\",\"isSidechain\":true,\"message\":{{\"role\":\"assistant\",\"content\":[{{\"type\":\"text\",\"text\":\"x\"}}]}},\"uuid\":\"{uuid}\"}}\n"
                ),
            )
            .unwrap();
        };
        write_sidecar("agent-older", "old");
        write_sidecar("agent-newer", "new");
        set_age(&sidecars.join("agent-older.meta.json"), 600);
        set_age(&sidecars.join("agent-newer.meta.json"), 60);
        let one = fs::metadata(sidecars.join("agent-newer.jsonl")).unwrap().len();

        let (sidecars, _) = subagent_sidecars_with_budget(session_id, &transcript, one);
        let uuids: Vec<_> =
            sidecars.iter().flat_map(|(_, rows)| rows.iter().map(|row| row.uuid.clone())).collect();

        assert!(uuids.iter().any(|uuid| uuid == "new"), "the newest sidecar is read");
        assert!(
            !uuids.iter().any(|uuid| uuid == "old"),
            "an older sidecar past the budget is skipped, not kept in place of the newest",
        );
    }

    /// A resumed git worker anchors its history read at the worktree run
    /// dir (`claude --resume` receives that cwd via `worker_tag_dir`), so
    /// the JSONL is read from the worktree project key, not the repo-root
    /// key. The same session id under the repo-root key is invisible -
    /// which is why the resume-map must pick a worktree-scoped session.
    #[test]
    fn get_session_messages_reads_git_worker_from_worktree_key() {
        let config_dir = tempfile::tempdir().unwrap();
        let session_id = "550e8400-e29b-41d4-a716-446655440000";
        let repo_root = "/Users/me/Projects/playground";
        let worktree = "/Users/me/Projects/playground/.claude/worktrees/gpt-tutor";

        let worktree_dir = project_dir_for(config_dir.path(), worktree);
        fs::create_dir_all(&worktree_dir).unwrap();
        write_session_jsonl(
            &worktree_dir,
            session_id,
            "{\"type\":\"user\",\"message\":{\"content\":\"hello from the worktree\"}}\n",
        );

        let from_worktree =
            get_session_messages(config_dir.path(), session_id, Some(worktree)).messages;
        assert_eq!(from_worktree.len(), 1, "resume reads the JSONL under the worktree key");

        let from_repo_root =
            get_session_messages(config_dir.path(), session_id, Some(repo_root)).messages;
        assert!(
            from_repo_root.is_empty(),
            "the same session id under the repo-root key is not where a git worker reads",
        );
    }

    /// A transcript of `rows` plain turns, each a user row carrying its own
    /// number and a uuid a caller can anchor on.
    fn a_transcript(rows: usize) -> String {
        use std::fmt::Write as _;
        let mut out = String::new();
        for at in 0..rows {
            let _ = writeln!(
                out,
                "{{\"type\":\"user\",\"uuid\":\"u{at}\",\"session_id\":\"s1\",\
                 \"message\":{{\"role\":\"user\",\"content\":\"turn {at}\"}}}}"
            );
        }
        out
    }

    /// An anchor on a row the file carries, without a byte: what a caller
    /// holding the row but not its position reads by.
    fn anchored(row: &str) -> forge_primitives::TranscriptAnchor {
        forge_primitives::TranscriptAnchor { row: row.to_owned(), offset: None }
    }

    /// The words a frame carries, for a span whose rows are its turns' names.
    fn said(message: &forge_primitives::Message) -> String {
        let forge_primitives::Message::User { message, .. } = message else {
            return String::new();
        };
        message
            .content
            .iter()
            .find_map(|block| match block {
                forge_primitives::ContentBlock::Text { text, .. } => Some(text.clone()),
                _ => None,
            })
            .unwrap_or_default()
    }

    /// The read's rows, in the words the file wrote them with.
    fn words(span: &forge_primitives::TranscriptSpan) -> Vec<String> {
        span.messages.iter().map(said).collect()
    }

    /// **A page reads the rows just below its anchor, down to the file's
    /// start.** The anchor names the row the page above stopped at; the rows
    /// the read serves are the ones strictly older than it, and the file's own
    /// first row is where the span says the history ends.
    #[test]
    fn a_row_read_serves_the_rows_below_its_anchor() {
        let config = tempfile::tempdir().unwrap();
        let cwd = "/Users/me/Projects/playground";
        let dir = project_dir_for(config.path(), cwd);
        fs::create_dir_all(&dir).unwrap();
        let session = "550e8400-e29b-41d4-a716-446655440000";
        write_session_jsonl(&dir, session, &a_transcript(40));

        let span = read_rows(config.path(), session, Some(cwd), Some(&anchored("u30")), 100)
            .expect("the transcript has rows below the anchor");

        assert!(span.exhausted, "the read reached the file's own first row");
        assert_eq!(span.messages.len(), 30, "every row below the anchor, and none above");
        assert_eq!(span.messages.len(), span.offsets.len(), "each row with the byte it starts at");
        let expected: Vec<String> = (0..30).map(|at| format!("turn {at}")).collect();
        assert_eq!(words(&span), expected, "in conversation order");

        // A page asks for its own budget of rows and gets the NEWEST of the
        // older rows: what does not fit is left for the page below it.
        let trimmed = read_rows(config.path(), session, Some(cwd), Some(&anchored("u30")), 5)
            .expect("a span the budget trims");
        assert_eq!(
            words(&trimmed),
            (25..30).map(|at| format!("turn {at}")).collect::<Vec<_>>(),
            "the page's own rows and no more, taken nearest the anchor",
        );
        assert!(!trimmed.exhausted, "which is not the file's start");

        // And the walk's first ask, on a window that holds no row to name: the
        // file's own newest rows.
        let newest = read_rows(config.path(), session, Some(cwd), None, 5)
            .expect("the file answers with its newest rows");
        assert_eq!(
            words(&newest),
            (35..40).map(|at| format!("turn {at}")).collect::<Vec<_>>(),
            "the newest rows the session wrote",
        );
        assert!(!newest.exhausted, "with more above them");
    }

    /// **A page reads the duplicated block once too.** The paging span read is
    /// the second reader of the same file, and it serves the rows a page below
    /// the window's floor draws - which is where a re-appended block sits. Its
    /// row-level dedupe keys on the `uuid` the same way the whole-file read's
    /// does.
    #[test]
    fn a_row_read_reads_a_duplicated_block_once() {
        use std::fmt::Write as _;
        let config = tempfile::tempdir().unwrap();
        let cwd = "/Users/me/Projects/playground";
        let dir = project_dir_for(config.path(), cwd);
        fs::create_dir_all(&dir).unwrap();
        let session = "550e8400-e29b-41d4-a716-446655440000";
        let mut body = a_transcript(6);
        // A later run's session-state rows, then the re-appended tail block:
        // u3..u5 written a second time under their original uuids.
        let _ = writeln!(body, "{{\"type\":\"last-prompt\",\"uuid\":\"lp1\"}}");
        for at in 3..6 {
            let _ = writeln!(
                body,
                "{{\"type\":\"user\",\"uuid\":\"u{at}\",\"slug\":\"s\",\"session_id\":\"s1\",\
                 \"message\":{{\"role\":\"user\",\"content\":\"turn {at}\"}}}}"
            );
        }
        write_session_jsonl(&dir, session, &body);

        let span = read_rows(config.path(), session, Some(cwd), None, 100)
            .expect("the transcript answers with its rows");
        assert_eq!(
            ids(&span),
            ["u0", "u1", "u2", "u3", "u4", "u5"],
            "the newest rows, the duplicated tail block among them only once",
        );
    }

    /// **A chained read seeks where a locating read searches.** Given the byte
    /// its anchor row starts at, the read takes the window ending there; given
    /// only the id, it has to find the row, which a window too small to reach
    /// it cannot do - and the walk's first page is the one that pays for the
    /// search.
    #[test]
    fn a_row_read_seeks_when_its_anchor_carries_a_byte() {
        // A cap of a couple of hundred bytes: two rows or so, which is not
        // where an anchor twenty rows back is.
        const SMALL: u64 = 200;
        let config = tempfile::tempdir().unwrap();
        let cwd = "/Users/me/Projects/playground";
        let dir = project_dir_for(config.path(), cwd);
        fs::create_dir_all(&dir).unwrap();
        let session = "550e8400-e29b-41d4-a716-446655440000";
        write_session_jsonl(&dir, session, &a_transcript(40));

        let located = read_rows(config.path(), session, Some(cwd), Some(&anchored("u30")), 100)
            .expect("the first read locates the anchor");
        let oldest = located.offsets.first().copied().expect("the span's oldest row byte");
        let oldest_row = said(located.messages.first().expect("a first"));
        assert_eq!(oldest_row, "turn 0", "precondition: the span ran to the file's first row");

        // The page below the span's own start: the next read is handed that
        // row's byte, and reads below it - which is nothing, because the span
        // held the file's start.
        let resume =
            forge_primitives::TranscriptAnchor { row: "u0".to_owned(), offset: Some(oldest) };
        let chained =
            read_rows_with(config.path(), session, Some(cwd), Some(&resume), 100, SMALL, SMALL)
                .expect("the resumed read seeks to the row's byte");
        assert!(chained.messages.is_empty(), "nothing sits below the file's first row");
        assert!(chained.exhausted, "and the read says so");

        // The same anchor without its byte cannot be reached inside the cap:
        // the search starts at the file's end and this cap does not reach
        // twenty rows back, so the read answers nothing rather than a page it
        // cannot vouch for.
        assert!(
            read_rows_with(
                config.path(),
                session,
                Some(cwd),
                Some(&anchored("u0")),
                100,
                SMALL,
                SMALL,
            )
            .is_none(),
            "a locating read whose window cannot reach its anchor answers nothing",
        );

        // And a chained read with a byte serves the rows just below it with no
        // search at all: under the same unreachably small cap, the read seeks
        // rather than searches and answers a short page with a cursor instead
        // of nothing.
        let chained = read_rows_with(
            config.path(),
            session,
            Some(cwd),
            Some(&forge_primitives::TranscriptAnchor {
                row: "u20".to_owned(),
                offset: located.offsets.get(20).copied(),
            }),
            100,
            SMALL,
            SMALL,
        )
        .expect("the chained read seeks to the row's byte");
        assert!(!chained.messages.is_empty(), "and answers the rows the byte reaches");
        assert_eq!(
            said(chained.messages.last().expect("a last")),
            "turn 19",
            "the newest row below the anchor",
        );
        assert!(!chained.exhausted, "with the file's start still above it");
    }

    /// A transcript of `pairs` tool calls, each an assistant row calling and
    /// the user row carrying its result.
    fn a_called_transcript(pairs: usize) -> String {
        use std::fmt::Write as _;
        let mut out = String::new();
        for at in 0..pairs {
            let _ = writeln!(
                out,
                "{{\"type\":\"assistant\",\"uuid\":\"call{at}\",\"session_id\":\"s1\",\
                 \"message\":{{\"id\":\"m{at}\",\"role\":\"assistant\",\"model\":\"claude-opus-5\",\
                 \"content\":[{{\"type\":\"tool_use\",\
                 \"id\":\"tu{at}\",\"name\":\"Bash\",\"input\":{{\"command\":\"ls\"}}}}]}}}}"
            );
            let _ = writeln!(
                out,
                "{{\"type\":\"user\",\"uuid\":\"result{at}\",\"session_id\":\"s1\",\
                 \"message\":{{\"role\":\"user\",\"content\":[{{\"type\":\"tool_result\",\
                 \"tool_use_id\":\"tu{at}\",\"content\":\"ok\"}}]}}}}"
            );
        }
        out
    }

    /// The ids of a span's rows, for a transcript whose rows each carry one.
    fn ids(span: &forge_primitives::TranscriptSpan) -> Vec<String> {
        span.messages
            .iter()
            .map(|message| match message {
                forge_primitives::Message::Assistant { uuid: Some(uuid), .. }
                | forge_primitives::Message::User { uuid: Some(uuid), .. }
                | forge_primitives::Message::StopHookSummary { uuid, .. }
                | forge_primitives::Message::CompactBoundary { uuid, .. } => uuid.clone(),
                other => panic!("every row of this transcript carries an id: {other:?}"),
            })
            .collect()
    }

    /// **A page never opens on a result without its call.** The rows a page
    /// keeps are the newest of the older rows, so a cut can land among result
    /// rows - and those are deferred to the page below, which is cut from the
    /// call that produced them. Nothing is dropped: the result the cut stepped
    /// past is carried with its call.
    #[test]
    fn a_row_read_defers_leading_results_to_the_page_below() {
        let config = tempfile::tempdir().unwrap();
        let cwd = "/Users/me/Projects/playground";
        let dir = project_dir_for(config.path(), cwd);
        fs::create_dir_all(&dir).unwrap();
        let session = "550e8400-e29b-41d4-a716-446655440000";
        write_session_jsonl(&dir, session, &a_called_transcript(20));

        // Three rows below the anchor land mid-pair, and the cut's room reaches
        // past that: the page opens on a CALL - the leading result rows are
        // deferred - and closes on the newest pair below the anchor.
        let span = read_rows(config.path(), session, Some(cwd), Some(&anchored("call10")), 3)
            .expect("a span below the anchor");
        assert_eq!(
            ids(&span),
            vec!["call9", "result9"],
            "the page opens on the call the cut stepped back to, rather than on a result without \
             its call",
        );

        // **The walk down, page after page.** Every page opens on a call, and
        // together they serve every row below the anchor exactly once, in
        // order - the rows a cut stepped past are the page below's newest.
        let mut seen: Vec<String> = Vec::new();
        let mut anchor = Some(anchored("call10"));
        let mut pages = 0_usize;
        while let Some(at) = anchor {
            let span = read_rows(config.path(), session, Some(cwd), Some(&at), 3).expect("a page");
            let page = ids(&span);
            assert!(
                page.first().is_some_and(|id| id.starts_with("call")),
                "every page below the floor opens on a call, never on a bare result: {page:?}",
            );
            assert!(page.len() <= 3, "and carries the rows it was asked for: {page:?}");
            seen.extend(page.clone());
            pages += 1;
            assert!(pages < 40, "the walk reaches the file's first row");
            anchor = (!span.exhausted).then(|| forge_primitives::TranscriptAnchor {
                row: page[0].clone(),
                offset: span.offsets.first().copied(),
            });
        }
        let expected: Vec<String> =
            (0..10).rev().flat_map(|at| [format!("call{at}"), format!("result{at}")]).collect();
        assert_eq!(
            seen, expected,
            "the whole span below the anchor, once each, newest page first, with every result \
             served beside the call that produced it",
        );
    }

    /// **A stale byte does not answer with a stale page.** The anchor's byte
    /// is verified before it is believed: a transcript rewritten under a walk
    /// carries another row at that byte, and the read falls back to searching
    /// for the anchor by id - so both routes answer the rows the id names.
    #[test]
    fn a_row_read_falls_back_when_the_anchors_byte_moved() {
        let config = tempfile::tempdir().unwrap();
        let cwd = "/Users/me/Projects/playground";
        let dir = project_dir_for(config.path(), cwd);
        fs::create_dir_all(&dir).unwrap();
        let session = "550e8400-e29b-41d4-a716-446655440000";
        write_session_jsonl(&dir, session, &a_transcript(40));

        let located = read_rows(config.path(), session, Some(cwd), Some(&anchored("u30")), 100)
            .expect("the first read locates the anchor");
        let last = located.messages.len() - 1;
        let resume = forge_primitives::TranscriptAnchor {
            row: "u29".to_owned(),
            offset: located.offsets.get(last).copied(),
        };
        assert!(resume.offset.is_some(), "precondition: the span gave the row its byte");

        // A row prepended: every byte below it moved, so the anchored byte no
        // longer holds the row it was read for.
        let prepended = format!(
            "{{\"type\":\"user\",\"uuid\":\"u_pre\",\"session_id\":\"s1\",\
             \"message\":{{\"role\":\"user\",\"content\":\"prepended\"}}}}\n{}",
            a_transcript(40),
        );
        write_session_jsonl(&dir, session, &prepended);

        let stale = read_rows(config.path(), session, Some(cwd), Some(&resume), 100);
        let located_again =
            read_rows(config.path(), session, Some(cwd), Some(&anchored("u29")), 100);
        assert_eq!(
            stale.map(|span| words(&span)),
            located_again.map(|span| words(&span)),
            "a stale byte answers exactly as the read that never had one: the anchor's row is \
             found again by its id, so the page below it is the page below it",
        );
    }

    /// **A sub-agent's frame has no transcript row.** The file's rule drops
    /// it, and a numbering that counted it as a row would be off for every
    /// page below it: the two halves have to say the same thing.
    #[test]
    fn a_sub_agents_frame_has_no_row() {
        let row = serde_json::json!({
            "type": "user",
            "uuid": "u1",
            "parent_tool_use_id": "toolu_dispatch",
            "session_id": "s1",
            "message": {"role": "user", "content": "under a sub-agent"},
        });
        let frame: forge_primitives::Message =
            serde_json::from_value(row.clone()).expect("a frame");
        assert!(!has_a_transcript_row(&frame), "a sub-agent's frame carries no row");
        assert!(transcript_row(&row).is_none(), "and the row rule drops it, as this says it does");
    }

    /// **A stamped refusal has no transcript row either.** Stamping a
    /// refusal with the dispatch it belongs under makes it a sub-agent frame
    /// the row rule drops, so a numbering that counted it as a row would be
    /// off for every page below it.
    #[test]
    fn a_stamped_refusal_has_no_row() {
        let row = serde_json::json!({
            "type": "system",
            "subtype": "permission_denied",
            "tool_name": "Bash",
            "tool_use_id": "call_refused",
            "decision_reason_type": "subcommandResults",
            "message": "Permission to use Bash has been denied.",
            "uuid": "p1",
            "session_id": "s1",
            "parent_tool_use_id": "toolu_dispatch",
        });
        let frame: forge_primitives::Message =
            serde_json::from_value(row).expect("a stamped refusal");
        assert!(!has_a_transcript_row(&frame), "a stamped refusal carries no row");
        let unstamped: forge_primitives::Message = serde_json::from_value(serde_json::json!({
            "type": "system",
            "subtype": "permission_denied",
            "tool_name": "Bash",
            "tool_use_id": "call_refused",
            "decision_reason_type": "other",
            "decision_reason": "Contains simple_expansion",
            "message": "Contains simple_expansion",
            "uuid": "p2",
            "session_id": "s1",
        }))
        .expect("a session-level refusal");
        assert!(has_a_transcript_row(&unstamped), "the session's own refusal keeps its row");
    }

    /// The one row rule's forks reach a span: a queued prompt's attachment row
    /// is hoisted into the block a view reads it as, and a boundary row is the
    /// frame it is - not skipped, and not flattened into something else.
    #[test]
    fn a_row_read_types_the_rows_the_conversion_has_forks_for() {
        let config = tempfile::tempdir().unwrap();
        let cwd = "/Users/me/Projects/playground";
        let dir = project_dir_for(config.path(), cwd);
        fs::create_dir_all(&dir).unwrap();
        let session = "550e8400-e29b-41d4-a716-446655440000";
        let body = format!(
            "{}\n{}\n{}\n{}\n{}",
            "{\"type\":\"user\",\"uuid\":\"u0\",\"session_id\":\"s1\",\
             \"message\":{\"role\":\"user\",\"content\":\"turn 0\"}}",
            "{\"type\":\"attachment\",\"uuid\":\"a1\",\"session_id\":\"s1\",\
             \"attachment\":{\"type\":\"queued_command\",\"prompt\":\"queued\",\"commandMode\":\"prompt\"}}",
            "{\"type\":\"queue-operation\",\"operation\":\"remove\",\"timestamp\":\"2026-11-03T12:00:00Z\",\
             \"sessionId\":\"s1\",\"content\":\"queued\"}",
            "{\"type\":\"system\",\"subtype\":\"compact_boundary\",\"uuid\":\"h1\",\
             \"sessionId\":\"s1\",\"compactMetadata\":{\"trigger\":\"auto\",\"preTokens\":1,\"postTokens\":1}}",
            "{\"type\":\"user\",\"uuid\":\"u2\",\"session_id\":\"s1\",\
             \"message\":{\"role\":\"user\",\"content\":\"turn 1\"}}",
        );
        write_session_jsonl(&dir, session, &body);

        let span = read_rows(config.path(), session, Some(cwd), Some(&anchored("u2")), 100)
            .expect("a span over the forks");

        assert!(span.exhausted, "every row the read kept is below the anchor");
        assert_eq!(span.messages.len(), 3, "the attachment's row is a frame, not a skip");
        assert!(
            matches!(
                &span.messages[1],
                forge_primitives::Message::User { message, .. }
                    if matches!(message.content[0], forge_primitives::ContentBlock::QueuedCommand { .. })
            ),
            "the queued command arrives as the block a view reads it as",
        );
        assert!(
            matches!(&span.messages[2], forge_primitives::Message::CompactBoundary { uuid, .. }
                if uuid == "h1"),
            "and the boundary row as the frame it is, not a skip: {:?}",
            span.messages[2],
        );
    }

    /// **Only a start reason draws the prompt.** A removal carrying any other
    /// reason did not take the prompt, so neither the removal nor the
    /// attachment it would have paired with draws a row.
    #[test]
    fn a_remove_reason_that_is_not_a_start_draws_no_prompt_row() {
        let jsonl = r#"{"type":"user","message":{"role":"user","content":"plain"},"uuid":"u1","session_id":"s1"}
{"type":"attachment","attachment":{"type":"queued_command","prompt":"cancelled words","commandMode":"prompt"},"uuid":"a1","session_id":"s1","timestamp":"2026-11-03T12:00:00Z"}
{"type":"queue-operation","operation":"remove","reason":"cancelled","timestamp":"2026-11-03T12:00:05Z","sessionId":"s1","content":"cancelled words"}
"#;
        let msgs = parse_session_messages(jsonl.as_bytes()).messages;
        assert_eq!(
            msgs.iter().map(|row| row.uuid.as_str()).collect::<Vec<_>>(),
            ["u1"],
            "the prompt a non-start removal names draws nothing",
        );
    }

    /// **The paged read holds a queued prompt until its start too.** An
    /// assistant row written between the attachment and its `remove` must
    /// come first in the served span: the page's order is the start's, and a
    /// span that served the attachment at its own byte would put the words
    /// above work the same turn wrote after the press.
    #[test]
    fn a_paged_read_serves_a_queued_prompt_at_its_remove() {
        let config = tempfile::tempdir().unwrap();
        let cwd = "/Users/me/Projects/playground";
        let dir = project_dir_for(config.path(), cwd);
        fs::create_dir_all(&dir).unwrap();
        let session = "550e8400-e29b-41d4-a716-446655440000";
        let body = format!(
            "{}\n{}\n{}\n{}\n{}",
            "{\"type\":\"user\",\"uuid\":\"u1\",\"session_id\":\"s1\",\
             \"message\":{\"role\":\"user\",\"content\":\"before\"},\
             \"timestamp\":\"2026-11-03T11:59:00Z\"}",
            "{\"type\":\"attachment\",\"uuid\":\"a1\",\"session_id\":\"s1\",\
             \"attachment\":{\"type\":\"queued_command\",\"prompt\":\"queued words\",\
             \"commandMode\":\"prompt\",\"source_uuid\":\"p1\"},\
             \"timestamp\":\"2026-11-03T12:00:00Z\"}",
            "{\"type\":\"assistant\",\"uuid\":\"as1\",\"session_id\":\"s1\",\
             \"message\":{\"id\":\"m1\",\"role\":\"assistant\",\"model\":\"claude-opus-5\",\
             \"content\":[{\"type\":\"text\",\"text\":\"working\"}]},\
             \"timestamp\":\"2026-11-03T12:00:01Z\"}",
            "{\"type\":\"queue-operation\",\"operation\":\"remove\",\
             \"reason\":\"absorbed_mid_turn\",\"timestamp\":\"2026-11-03T12:00:05Z\",\
             \"sessionId\":\"s1\",\"content\":\"queued words\"}",
            "{\"type\":\"user\",\"uuid\":\"u2\",\"session_id\":\"s1\",\
             \"message\":{\"role\":\"user\",\"content\":\"after\"},\
             \"timestamp\":\"2026-11-03T12:00:06Z\"}",
        );
        write_session_jsonl(&dir, session, &body);

        let span = read_rows(config.path(), session, Some(cwd), Some(&anchored("u2")), 100)
            .expect("a span over the queued prompt");

        assert_eq!(span.messages.len(), 3, "the page carries the rows below the anchor");
        assert!(
            matches!(&span.messages[1], forge_primitives::Message::Assistant { .. }),
            "the assistant row written between the press and the start is served first",
        );
        let forge_primitives::Message::User { uuid, timestamp, message, .. } = &span.messages[2]
        else {
            panic!("the queued prompt is a user row: {:?}", span.messages[2]);
        };
        assert!(
            uuid.as_deref().is_some_and(|id| id.starts_with("queue-remove-2026-11-03T12:00:05Z-")),
            "named by the start it was served at: {uuid:?}",
        );
        assert_eq!(
            timestamp.as_deref(),
            Some("2026-11-03T12:00:05Z"),
            "stamped at the start, not the press",
        );
        assert!(
            message.content.iter().any(|block| matches!(
                block,
                forge_primitives::ContentBlock::QueuedCommand { prompt, source_uuid, .. }
                    if prompt.as_str() == Some("queued words")
                        && source_uuid.as_deref() == Some("p1")
            )),
            "carrying the prompt's words and the id the live row pairs with: {message:?}",
        );
    }

    /// **The read is a window, not the file.** The rows are found in a tail
    /// window, and it grows only while the rows asked for are not inside it,
    /// so the transcript's older megabytes are not read for a page that lands
    /// inside the first window.
    #[test]
    fn a_row_read_grows_its_window_only_while_the_rows_are_outside_it() {
        let config = tempfile::tempdir().unwrap();
        let cwd = "/Users/me/Projects/playground";
        let dir = project_dir_for(config.path(), cwd);
        fs::create_dir_all(&dir).unwrap();
        let session = "550e8400-e29b-41d4-a716-446655440000";
        let body = a_transcript(40);
        write_session_jsonl(&dir, session, &body);
        let len = body.len() as u64;

        // A file's worth of cap, a tenth of it a step: the first window holds
        // the newest rows and not the thirty a page asked for, so the read
        // grows - and the page it answers is the one budgeted, not the window.
        let grown = read_rows_with(config.path(), session, Some(cwd), None, 30, len / 10, len)
            .expect("the read grows to hold the page");
        assert_eq!(grown.messages.len(), 30, "the page's own rows and no more");
        assert_eq!(
            said(grown.messages.first().expect("a first")),
            "turn 10",
            "taken nearest the boundary",
        );
        assert!(!grown.exhausted, "which is not the file's start");

        // And the cap is where the search stops: an anchor further from the
        // end than the cap reaches answers nothing rather than a long read.
        assert!(
            read_rows_with(
                config.path(),
                session,
                Some(cwd),
                Some(&anchored("u5")),
                100,
                len / 5,
                len / 5,
            )
            .is_none(),
            "an anchor the cap cannot reach is not answered by reading further",
        );
    }

    /// An anchor the file does not carry is a page the read cannot vouch for:
    /// the search finds no such row and the answer is nothing, which the
    /// caller draws as the end of the history.
    #[test]
    fn a_row_read_refuses_an_anchor_the_file_does_not_carry() {
        let config = tempfile::tempdir().unwrap();
        let cwd = "/Users/me/Projects/playground";
        let dir = project_dir_for(config.path(), cwd);
        fs::create_dir_all(&dir).unwrap();
        let session = "550e8400-e29b-41d4-a716-446655440000";
        write_session_jsonl(&dir, session, &a_transcript(40));

        assert!(
            read_rows(config.path(), session, Some(cwd), Some(&anchored("nowhere")), 100).is_none(),
            "an anchor the file does not carry answers nothing",
        );
    }

    /// No file for the session is no page: the read answers nothing rather
    /// than an empty span a caller could read as the end of a history.
    #[test]
    fn a_row_read_answers_nothing_without_a_transcript() {
        let config = tempfile::tempdir().unwrap();
        let cwd = "/Users/me/Projects/playground";
        let dir = project_dir_for(config.path(), cwd);
        fs::create_dir_all(&dir).unwrap();
        let session = "550e8400-e29b-41d4-a716-446655440000";

        assert!(
            read_rows(config.path(), session, Some(cwd), Some(&anchored("u1")), 100).is_none(),
            "no file for the session is no span",
        );
        write_session_jsonl(&dir, session, "");
        assert!(
            read_rows(config.path(), session, Some(cwd), None, 100).is_none(),
            "and an empty file is none either",
        );
        assert!(
            read_rows(config.path(), "../escape", Some(cwd), None, 100).is_none(),
            "and an id that is not a uuid names a path this refuses rather than resolves",
        );
    }
}