lifeloop-cli 0.3.0

//! Codex CLI lifecycle telemetry reader.
//!
//! Parses Codex session JSONL emitting `event_msg` / `token_count`
//! payloads and extracts the lifecycle kernel: latest prompt-token
//! count, model name, and the natively reported context window.

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

use serde::Deserialize;
use serde_json::Value;

use super::{
    EnvAlias, PressureObservation, TelemetryError, TelemetryResult, TokenUsage, compute_pct,
    file_mtime_epoch_s, general_context_window, general_host_model, home_dir, read_file_bounded,
    resolve_env_string, string_key,
};

const ADAPTER_ID: &str = "codex";
const MAX_SESSION_SCAN_DEPTH: usize = 8;
const MAX_SESSION_SCAN_ENTRIES: usize = 10_000;

const CODEX_HOME_ALIASES: &[EnvAlias] = &[EnvAlias {
    lifeloop: "LIFELOOP_CODEX_HOME",
    ccd_compat: "CODEX_HOME",
}];

const CODEX_THREAD_ID_ALIASES: &[EnvAlias] = &[EnvAlias {
    lifeloop: "LIFELOOP_CODEX_THREAD_ID",
    ccd_compat: "CODEX_THREAD_ID",
}];

const CODEX_MODEL_ALIASES: &[EnvAlias] = &[EnvAlias {
    lifeloop: "LIFELOOP_CODEX_MODEL",
    ccd_compat: "CCD_CODEX_MODEL",
}];

/// Probe the active Codex session log for the configured thread id.
/// Returns `Ok(None)` when no thread id is configured or the session
/// log has no `token_count` events yet.
pub fn current() -> TelemetryResult<Option<PressureObservation>> {
    let Some(thread_id) = resolve_env_string(CODEX_THREAD_ID_ALIASES) else {
        return Ok(None);
    };

    let codex_home = codex_home()?;
    let Some(session_path) = find_session_log(&codex_home.join("sessions"), &thread_id)? else {
        return Ok(None);
    };

    let observed_at_epoch_s = match file_mtime_epoch_s(&session_path)? {
        Some(epoch_s) => epoch_s,
        None => return Ok(None),
    };

    let bytes = read_file_bounded(&session_path, "Codex session log")?;
    parse_session_log(&bytes, observed_at_epoch_s)
}

/// Parse a Codex session log byte slice. Public for tests and for
/// callers that read the file themselves.
pub fn parse_session_log(
    bytes: &[u8],
    observed_at_epoch_s: u64,
) -> TelemetryResult<Option<PressureObservation>> {
    let reader = BufReader::new(bytes);
    let mut latest: Option<PressureObservation> = None;
    let mut latest_model = resolve_env_string(CODEX_MODEL_ALIASES).or_else(general_host_model);

    for line in reader.lines() {
        let line = line.map_err(TelemetryError::from)?;
        let value: Value = match serde_json::from_str(&line) {
            Ok(v) => v,
            Err(_) => continue,
        };
        if let Some(model_name) = string_key(
            &value,
            &[
                "model",
                "model_name",
                "modelName",
                "model_slug",
                "modelSlug",
            ],
        ) {
            latest_model = Some(model_name);
        }
        let entry: SessionEntry = match serde_json::from_value(value) {
            Ok(e) => e,
            Err(_) => continue,
        };
        if entry.entry_type != "event_msg" {
            continue;
        }
        let Some(payload) = entry.payload else {
            continue;
        };
        if payload.payload_type != "token_count" {
            continue;
        }

        let native_window = match payload.info.model_context_window {
            0 => None,
            value => Some(value),
        };
        let context_window = native_window.or_else(general_context_window);

        let total_tokens = payload
            .info
            .last_token_usage
            .as_ref()
            .map(|u| u.total_tokens)
            .unwrap_or(payload.info.total_token_usage.total_tokens);

        latest = Some(PressureObservation {
            adapter_id: ADAPTER_ID.into(),
            adapter_version: None,
            observed_at_epoch_s,
            model_name: latest_model.clone(),
            total_tokens: Some(total_tokens),
            context_window_tokens: context_window,
            context_used_pct: compute_pct(total_tokens, context_window),
            compaction_signal: None,
            usage: TokenUsage {
                blended_total_tokens: Some(payload.info.total_token_usage.total_tokens),
                ..TokenUsage::default()
            },
        });
    }

    Ok(latest)
}

fn codex_home() -> TelemetryResult<PathBuf> {
    if let Some(path) = resolve_env_string(CODEX_HOME_ALIASES) {
        return Ok(PathBuf::from(path));
    }
    Ok(home_dir()?.join(".codex"))
}

fn find_session_log(root: &Path, thread_id: &str) -> TelemetryResult<Option<PathBuf>> {
    let root_meta = match fs::symlink_metadata(root) {
        Ok(meta) => meta,
        Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(None),
        Err(e) => return Err(TelemetryError::from(e)),
    };
    if root_meta.file_type().is_symlink() || !root_meta.is_dir() {
        return Ok(None);
    }
    let mut visited = HashSet::new();
    let mut stack = vec![(root.to_path_buf(), 0usize)];
    let mut entries_seen = 0usize;
    while let Some((dir, depth)) = stack.pop() {
        let canonical = fs::canonicalize(&dir).map_err(TelemetryError::from)?;
        if !visited.insert(canonical) {
            continue;
        }
        for entry in fs::read_dir(&dir).map_err(TelemetryError::from)? {
            let entry = entry.map_err(TelemetryError::from)?;
            entries_seen += 1;
            if entries_seen > MAX_SESSION_SCAN_ENTRIES {
                return Err(TelemetryError::Unavailable(format!(
                    "Codex session search exceeded {MAX_SESSION_SCAN_ENTRIES} entries under {}",
                    root.display()
                )));
            }
            let path = entry.path();
            let file_type = entry.file_type().map_err(TelemetryError::from)?;
            if file_type.is_symlink() {
                continue;
            }
            if file_type.is_dir() {
                if depth < MAX_SESSION_SCAN_DEPTH {
                    stack.push((path, depth + 1));
                }
                continue;
            }
            if !file_type.is_file() {
                continue;
            }
            let matches = path
                .file_name()
                .and_then(|n| n.to_str())
                .map(|n| n.contains(thread_id) && n.ends_with(".jsonl"))
                .unwrap_or(false);
            if matches {
                return Ok(Some(path));
            }
        }
    }
    Ok(None)
}

#[derive(Deserialize)]
struct SessionEntry {
    #[serde(rename = "type")]
    entry_type: String,
    payload: Option<TokenCountPayload>,
}

#[derive(Deserialize)]
struct TokenCountPayload {
    #[serde(rename = "type")]
    payload_type: String,
    info: TokenCountInfo,
}

#[derive(Deserialize)]
struct TokenCountInfo {
    total_token_usage: TotalTokenUsage,
    last_token_usage: Option<TotalTokenUsage>,
    #[serde(default)]
    model_context_window: u64,
}

#[derive(Deserialize)]
struct TotalTokenUsage {
    total_tokens: u64,
}

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

    #[cfg(unix)]
    #[test]
    fn find_session_log_skips_symlink_directories() {
        let root = tempfile::tempdir().expect("temp root");
        let outside = tempfile::tempdir().expect("temp outside");
        fs::write(outside.path().join("thread-123.jsonl"), "{}\n").expect("outside log");
        std::os::unix::fs::symlink(outside.path(), root.path().join("linked"))
            .expect("symlink dir");

        let found = find_session_log(root.path(), "thread-123").expect("search ok");
        assert!(
            found.is_none(),
            "symlinked session dirs must not be followed"
        );
    }

    #[test]
    fn find_session_log_finds_regular_nested_file() {
        let root = tempfile::tempdir().expect("temp root");
        let nested = root.path().join("2026").join("05");
        fs::create_dir_all(&nested).expect("nested dirs");
        let expected = nested.join("thread-abc.jsonl");
        fs::write(&expected, "{}\n").expect("session log");

        let found = find_session_log(root.path(), "thread-abc").expect("search ok");
        assert_eq!(found.as_deref(), Some(expected.as_path()));
    }
}