openlatch-client 0.6.23

//! The agent's version, read from the session file its hook payload names.
//!
//! Neither Codex's nor Claude Code's hook payloads carry a version, so
//! `agentversion` went out NULL on every row of theirs, and a host running a
//! Codex too old to know some of our hook events could not be found from the
//! data. Both payloads carry `transcript_path`, and both files record the
//! version of the build that is writing them:
//!
//! - **a `session_meta` record** with `payload.cli_version`, Codex's first line
//!   (filled from the build version since at least 0.130);
//! - **an entry with `sessionId` and `version`**, Claude Code's — not its first
//!   line, but within its first seven on every transcript measured.
//!
//! Matched on those SHAPES, never on which agent sent the event. The first
//! [`MAX_SCAN_LINES`] lines and [`MAX_SCAN_BYTES`] bytes are read; a file that
//! passes either without naming a version answers nothing, and that answer is
//! remembered.
//!
//! The egress has a pure in-memory budget, so it only ever asks
//! [`version_for`], which looks in a cache. A miss starts one background read
//! and answers `None`: the session's later events carry the version. A file
//! not written yet, or still shorter than the scan with no version in it (a
//! `SessionStart` can fire first), is not remembered, so the next event asks
//! again.

use std::collections::{HashMap, HashSet};
use std::io::{BufRead as _, Read as _};
use std::path::{Component, Path};
use std::sync::{Mutex, OnceLock};

use serde_json::Value;

/// The most of a session file that is read. Codex's first line runs to about
/// 23 KB, because it embeds the instructions, and Claude Code's version entry
/// came within 42 KB on every transcript measured; this only bounds a hostile
/// or unexpected file.
const MAX_SCAN_BYTES: u64 = 1024 * 1024;

/// The most lines read looking for a version entry.
const MAX_SCAN_LINES: usize = 64;

/// Sessions remembered before the cache starts again. One entry per session
/// file, so this is days of sessions on a busy host.
const MAX_REMEMBERED: usize = 4096;

/// What one session file says about the agent's version.
#[derive(Debug, Clone, PartialEq, Eq)]
enum Answer {
    Version(String),
    /// Read, and not a session file of this shape. Never asked again.
    NotSessionMeta,
}

#[derive(Default)]
struct Cache {
    answers: HashMap<String, Answer>,
    reading: HashSet<String>,
}

fn cache() -> &'static Mutex<Cache> {
    static CACHE: OnceLock<Mutex<Cache>> = OnceLock::new();
    CACHE.get_or_init(Mutex::default)
}

/// The version the session file named by `data.transcript_path` records, when
/// it has already been read. Never blocks on the file: a miss schedules a read
/// on the blocking pool and answers `None`.
pub fn version_for(data: &Value) -> Option<String> {
    let path = data.get("transcript_path").and_then(Value::as_str)?;
    if !plausible_session_file(Path::new(path)) {
        return None;
    }
    let mut cache = cache().lock().unwrap_or_else(|e| e.into_inner());
    match cache.answers.get(path) {
        Some(Answer::Version(v)) => return Some(v.clone()),
        Some(Answer::NotSessionMeta) => return None,
        None => {}
    }
    let Ok(runtime) = tokio::runtime::Handle::try_current() else {
        return None;
    };
    if !cache.reading.insert(path.to_string()) {
        return None;
    }
    drop(cache);
    let path = path.to_string();
    runtime.spawn_blocking(move || {
        let answer = read_answer(Path::new(&path));
        remember(path, answer);
    });
    None
}

/// Record what reading `path` found; `None` (not written yet) is forgotten so
/// the next event asks again.
fn remember(path: String, answer: Option<Answer>) {
    let mut cache = cache().lock().unwrap_or_else(|e| e.into_inner());
    cache.reading.remove(&path);
    let Some(answer) = answer else {
        return;
    };
    if cache.answers.len() >= MAX_REMEMBERED {
        cache.answers.clear();
    }
    cache.answers.insert(path, answer);
}

/// The path is checked by structure before it is read, since it comes from the
/// payload: absolute, nothing but plain components, a `.jsonl` name.
fn plausible_session_file(path: &Path) -> bool {
    path.is_absolute()
        && path.components().all(|c| {
            matches!(
                c,
                Component::RootDir | Component::Prefix(_) | Component::Normal(_)
            )
        })
        && path.extension().is_some_and(|ext| ext == "jsonl")
}

/// `None` when the file cannot be read, or ends before the scan does without
/// naming a version (still being written), so it is asked again; otherwise
/// the version, or that the scan passed without one.
fn read_answer(path: &Path) -> Option<Answer> {
    // A regular file only: a FIFO named by a payload would park the reader.
    if !std::fs::metadata(path).ok()?.is_file() {
        return Some(Answer::NotSessionMeta);
    }
    let file = std::fs::File::open(path).ok()?;
    let mut reader = std::io::BufReader::new(file.take(MAX_SCAN_BYTES));
    let mut read = 0u64;
    let mut line = Vec::new();
    for _ in 0..MAX_SCAN_LINES {
        line.clear();
        let n = reader.read_until(b'\n', &mut line).ok()?;
        read += n as u64;
        if !line.ends_with(b"\n") {
            // The cap, or the end of what is written so far: a partial line
            // is never read as a record.
            return (read >= MAX_SCAN_BYTES).then_some(Answer::NotSessionMeta);
        }
        if let Some(version) = version_in(&line) {
            return Some(Answer::Version(version));
        }
    }
    Some(Answer::NotSessionMeta)
}

/// The version one session-file line records, by either shape, when it looks
/// like a version.
fn version_in(line: &[u8]) -> Option<String> {
    let record: Value = serde_json::from_slice(line).ok()?;
    let version = if record.get("type").and_then(Value::as_str) == Some("session_meta") {
        record.get("payload")?.get("cli_version")?.as_str()?
    } else {
        record.get("sessionId")?.as_str()?;
        record.get("version")?.as_str()?
    };
    let version = version.trim();
    crate::hooks::agent_versions::looks_like_version(version).then(|| version.to_string())
}

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

    fn session_file(dir: &Path, first_line: &str) -> String {
        let path = dir.join("rollout-2026-10-20T13-14-35-s.jsonl");
        std::fs::write(&path, format!("{first_line}\n{{\"type\":\"event_msg\"}}\n"))
            .expect("session file");
        path.display().to_string()
    }

    /// The shape Codex 0.162.1 wrote, trimmed to the fields read.
    const CODEX_META: &str = r#"{"timestamp":"2026-10-20T11:14:35Z","type":"session_meta","payload":{"id":"s","originator":"codex_exec","cli_version":"0.130.0","source":"exec","instructions":"…"}}"#;

    #[test]
    fn a_session_meta_first_line_names_the_version() {
        let dir = tempfile::tempdir().expect("dir");
        let path = session_file(dir.path(), CODEX_META);
        assert_eq!(
            read_answer(Path::new(&path)),
            Some(Answer::Version("0.130.0".into()))
        );
    }

    /// Claude Code's shape: not the first line, an entry with `sessionId`
    /// and `version` a few lines in — as its transcripts are written.
    #[test]
    fn a_claude_code_transcript_names_its_version_a_few_lines_in() {
        let dir = tempfile::tempdir().expect("dir");
        let path = dir.path().join("8f2c1e4a.jsonl");
        std::fs::write(
            &path,
            concat!(
                "{\"type\":\"last-prompt\",\"sessionId\":\"s\"}\n",
                "{\"type\":\"permission-mode\",\"sessionId\":\"s\"}\n",
                "{\"type\":\"attachment\",\"sessionId\":\"s\",\"version\":\"2.1.304\"}\n",
            ),
        )
        .expect("transcript");
        assert_eq!(read_answer(&path), Some(Answer::Version("2.1.304".into())));
        // A `version` with no `sessionId` beside it is some other record's.
        let other = dir.path().join("other.jsonl");
        std::fs::write(&other, "{\"version\":\"1.0.0\"}\n").expect("other");
        assert_eq!(read_answer(&other), None);
    }

    /// Anything else: no version. Asked again while the file is shorter than
    /// the scan (it may still be written), remembered once the scan passes.
    #[test]
    fn another_shape_answers_nothing_and_is_remembered_once_the_scan_passes() {
        let dir = tempfile::tempdir().expect("dir");
        let path = session_file(dir.path(), r#"{"type":"user","message":{"role":"user"}}"#);
        assert_eq!(read_answer(Path::new(&path)), None, "short: asked again");
        let long = dir.path().join("long.jsonl");
        std::fs::write(&long, "{\"type\":\"user\"}\n".repeat(MAX_SCAN_LINES + 1)).expect("long");
        assert_eq!(read_answer(&long), Some(Answer::NotSessionMeta));
        let junk = dir.path().join("junk.jsonl");
        std::fs::write(&junk, "not json at all\n".repeat(MAX_SCAN_LINES)).expect("junk");
        assert_eq!(read_answer(&junk), Some(Answer::NotSessionMeta));
    }

    /// A partial last line — the record being written — is never parsed.
    #[test]
    fn a_partial_line_is_not_read_as_a_record() {
        let dir = tempfile::tempdir().expect("dir");
        let path = dir.path().join("partial.jsonl");
        std::fs::write(
            &path,
            r#"{"type":"session_meta","payload":{"cli_version":"0.130.0"}}"#,
        )
        .expect("partial");
        assert_eq!(read_answer(&path), None);
    }

    /// Not written yet: asked again on the next event.
    #[test]
    fn a_file_not_written_yet_is_asked_again() {
        let dir = tempfile::tempdir().expect("dir");
        assert_eq!(read_answer(&dir.path().join("absent.jsonl")), None);
        let empty = dir.path().join("empty.jsonl");
        std::fs::write(&empty, "").expect("empty");
        assert_eq!(read_answer(&empty), None);
    }

    #[test]
    fn a_value_that_is_not_a_version_is_refused() {
        for bad in ["", "   ", "0.130.0; rm -rf /", &"9".repeat(65)] {
            let line = json!({"type": "session_meta", "payload": {"cli_version": bad}});
            assert_eq!(version_in(line.to_string().as_bytes()), None, "{bad:?}");
        }
        let line = json!({"type": "session_meta", "payload": {"cli_version": "0.164.0-alpha.6"}});
        assert_eq!(
            version_in(line.to_string().as_bytes()).as_deref(),
            Some("0.164.0-alpha.6")
        );
    }

    /// The first line past the cap is not a session_meta: nothing is guessed
    /// from a truncated record.
    #[test]
    fn an_oversized_first_line_answers_nothing() {
        let dir = tempfile::tempdir().expect("dir");
        let huge = format!(
            r#"{{"type":"session_meta","payload":{{"instructions":"{}","cli_version":"0.130.0"}}}}"#,
            "x".repeat(MAX_SCAN_BYTES as usize)
        );
        let path = session_file(dir.path(), &huge);
        assert_eq!(read_answer(Path::new(&path)), Some(Answer::NotSessionMeta));
    }

    #[test]
    fn a_path_from_the_payload_is_checked_before_it_is_read() {
        let dir = tempfile::tempdir().expect("dir");
        assert!(!plausible_session_file(Path::new("relative/s.jsonl")));
        assert!(!plausible_session_file(
            &dir.path().join("..").join("s.jsonl")
        ));
        assert!(!plausible_session_file(&dir.path().join("passwd")));
        assert!(plausible_session_file(&dir.path().join("rollout-s.jsonl")));
    }

    /// The egress never waits: the first ask schedules the read and answers
    /// nothing, a later one answers from the cache.
    #[tokio::test]
    async fn the_first_ask_schedules_and_a_later_one_answers() {
        let dir = tempfile::tempdir().expect("dir");
        let path = session_file(dir.path(), CODEX_META);
        let data = json!({"transcript_path": path, "hook_event_name": "PreToolUse"});

        assert_eq!(version_for(&data), None, "never blocks on the file");
        let mut answered = None;
        for _ in 0..200 {
            tokio::time::sleep(std::time::Duration::from_millis(5)).await;
            answered = version_for(&data);
            if answered.is_some() {
                break;
            }
        }
        assert_eq!(answered.as_deref(), Some("0.130.0"));
        assert_eq!(version_for(&json!({"tool_name": "Bash"})), None);
    }
}