csift 0.12.2

//! Transcript tail state machine: what the newest records say is happening.
//!
//! Reads the FINAL window of a transcript (bounded, never the whole file), walks it
//! backward, and reports the liveness-relevant shape: the newest UNRETURNED tool call
//! (a use whose id has no later result = a tool in flight, or a process dead mid-tool),
//! the last assistant `stop_reason`, and the last record's instant. A record is only
//! trusted from a COMPLETE line (torn tails are skipped by the newline framing).
//!
//! F9: growth alone is never activity - a main transcript grows while idle (enqueues,
//! attachments). This module classifies WHAT the tail is, not whether the file moved.

use super::*;

/// The bounded tail window in bytes: generous enough to cover any real turn tail
/// (hundreds of records), small enough to stay O(1) against a 300MB transcript.
const TAIL_WINDOW_BYTES: usize = 512 * 1024;

/// What a transcript's tail says right now.
#[derive(Debug, Clone, Default)]
pub(crate) struct TailShape {
    /// The newest tool_use whose id has NO later tool_result in the window: a tool in
    /// flight (or dead mid-tool; the pid probe disambiguates). `(tool name, use ts)`.
    pub(crate) unreturned_use: Option<(String, Option<String>)>,
    /// The newest assistant record's `stop_reason` (trustworthy on the MAIN lane;
    /// null is NORMAL mid-message on subagents).
    pub(crate) last_stop_reason: Option<String>,
    /// The newest record's timestamp (any type that carries one).
    pub(crate) last_ts_utc: Option<String>,
    /// Records inspected (evidence sizing; 0 = empty/unreadable file).
    pub(crate) records_seen: usize,
}

/// Read + classify the tail window of `path`.
pub(crate) fn tail_shape(path: &Path) -> Result<TailShape> {
    let mut shape = TailShape::default();
    // A plain positional read, never a memory map: this window is re-read on every
    // `wait` poll of a LIVE transcript, and Claude Code rewrites a transcript in place
    // (a rewind tombstone truncates and rewrites the tail), which would fault a map
    // with SIGBUS (v0.10.4).
    let (buf, start) = read_tail(path, TAIL_WINDOW_BYTES as u64)?;
    if buf.is_empty() {
        return Ok(shape);
    }
    let bytes: &[u8] = &buf;
    // Align to a line start (skip the partial line the cut landed in), unless we have
    // the whole file.
    let window = if start == 0 {
        bytes
    } else {
        match memchr::memchr(b'\n', bytes) {
            Some(nl) => &bytes[nl + 1..],
            None => &bytes[bytes.len()..],
        }
    };

    // Complete lines only: a torn final line (no trailing newline) is held, not parsed.
    let mut lines: Vec<&[u8]> = Vec::new();
    let mut pos = 0usize;
    while pos < window.len() {
        match memchr::memchr(b'\n', &window[pos..]) {
            Some(nl) => {
                lines.push(&window[pos..pos + nl]);
                pos += nl + 1;
            }
            None => break, // torn tail: skip (the next poll re-reads it complete)
        }
    }

    // Backward walk: result ids seen so far are LATER in file order, so a use id absent
    // from that set is unreturned as of the tail.
    let mut later_result_ids: std::collections::HashSet<String> = std::collections::HashSet::new();
    for line in lines.iter().rev() {
        let Ok(Some(rec)) = crate::parse::parse_line(line) else {
            continue;
        };
        shape.records_seen += 1;
        if shape.last_ts_utc.is_none() {
            shape.last_ts_utc = rec.timestamp.clone();
        }
        if shape.last_stop_reason.is_none() && rec.r#type.as_deref() == Some("assistant") {
            if let Some(sr) = rec.message.as_ref().and_then(|m| m.stop_reason.as_deref()) {
                shape.last_stop_reason = Some(sr.to_string());
            }
        }
        if let Some(blocks) = rec.blocks() {
            // Within one record, results precede later uses only across records; collect
            // results first so a same-record use+result (never real) stays paired.
            for b in blocks {
                if let crate::model::Block::ToolResult {
                    tool_use_id: Some(id),
                    ..
                } = b
                {
                    later_result_ids.insert(id.clone());
                }
            }
            if shape.unreturned_use.is_none() {
                for b in blocks {
                    if let crate::model::Block::ToolUse {
                        id: Some(id), name, ..
                    } = b
                    {
                        if !later_result_ids.contains(id) {
                            shape.unreturned_use = Some((
                                name.clone().unwrap_or_else(|| "(unnamed)".to_string()),
                                rec.timestamp.clone(),
                            ));
                        }
                    }
                }
            }
        }
        // Enough signal: stop once we have all three answers (bounded work even inside
        // the window).
        if shape.unreturned_use.is_some()
            && shape.last_stop_reason.is_some()
            && shape.records_seen >= 8
        {
            break;
        }
    }
    Ok(shape)
}

/// The line type the cost ledger writes as its checkpoint. It carries no `uuid`, no
/// `timestamp` and no `message`, so it is not a conversation record and the only
/// address it answers to is its physical line.
pub(crate) const CHECKPOINT_KIND: &str = "cost-state";

/// A cost-ledger checkpoint sitting at a transcript's very end.
#[derive(Debug, Clone)]
pub(crate) struct CheckpointTail {
    /// The checkpoint's 1-based physical line.
    pub(crate) line: usize,
    pub(crate) kind: &'static str,
}

/// The checkpoint at `path`'s tail, if its LAST non-blank line is one.
///
/// A checkpoint is written at a clear, a background handover, an in-app resume and at
/// exit, and it is repeatable - most of them sit mid-file - so finding one at the tail
/// says the harness stopped appending after writing it, and nothing more. It is
/// deliberately NOT a verdict: the seven-verdict set stays closed and the registry row
/// keeps deciding liveness.
pub(crate) fn last_checkpoint(path: &Path) -> Result<Option<CheckpointTail>> {
    static COST: std::sync::LazyLock<memchr::memmem::Finder<'static>> =
        std::sync::LazyLock::new(|| memchr::memmem::Finder::new(b"\"cost-state\""));
    let (buf, start) = read_tail(path, TAIL_WINDOW_BYTES as u64)?;
    let mut end = buf.len();
    while end > 0 && buf[end - 1].is_ascii_whitespace() {
        end -= 1;
    }
    if end == 0 {
        return Ok(None);
    }
    let line_start = memchr::memrchr(b'\n', &buf[..end]).map_or(0, |i| i + 1);
    // The window is bounded, so a last line that reaches its head may be cut: read
    // nothing rather than half a line.
    if line_start == 0 && start > 0 {
        return Ok(None);
    }
    let line = &buf[line_start..end];
    if COST.find(line).is_none() {
        return Ok(None);
    }
    let Ok(Some(rec)) = crate::parse::parse_line(line) else {
        return Ok(None);
    };
    if rec.r#type.as_deref() != Some(CHECKPOINT_KIND) {
        return Ok(None);
    }
    let at = start + line_start as u64;
    Ok(Some(CheckpointTail {
        line: count_newlines_before(path, at)? + 1,
        kind: CHECKPOINT_KIND,
    }))
}

/// Newlines in `path[..offset)`, read positionally in chunks - never a map, because
/// `wait` re-reads a live transcript (v0.10.4). Paid only when the tail IS a
/// checkpoint, which is the rare shape.
fn count_newlines_before(path: &Path, offset: u64) -> Result<usize> {
    const CHUNK: u64 = 4 * 1024 * 1024;
    let mut n = 0usize;
    let mut at = 0u64;
    while at < offset {
        let buf = read_range(path, at, (at + CHUNK).min(offset))?;
        if buf.is_empty() {
            break; // the file shrank under us: count what is there, never spin
        }
        n += memchr::memchr_iter(b'\n', &buf).count();
        at += buf.len() as u64;
    }
    Ok(n)
}

/// Seconds between an ISO instant and now; `None` when absent/unparseable.
pub(crate) fn age_secs(ts_utc: Option<&str>) -> Option<i64> {
    let t: jiff::Timestamp = ts_utc?.parse().ok()?;
    Some((jiff::Timestamp::now().as_second() - t.as_second()).max(0))
}