lifeloop/telemetry/

mod.rs

//! Lifecycle telemetry readers.
//!
//! Lifeloop owns lifecycle-relevant telemetry parsing for harness adapters.
//! Clients (CCD, RLM, and other lifecycle clients) consume neutral
//! [`PressureObservation`] snapshots instead of writing per-harness log
//! readers themselves.
//!
//! # Boundary (issue #5)
//!
//! This module owns:
//! * parsing harness-native session/telemetry artifacts into a neutral
//!   [`PressureObservation`] (adapter id, observed time, model name, token
//!   counts, context window, compaction signal),
//! * the env-var resolution rules used to locate those artifacts and to
//!   override their parsed values (with `LIFELOOP_*` aliases winning over
//!   the compatibility `CCD_*` inputs),
//! * a neutral telemetry [`TelemetryError`] type whose variants name the
//!   lifecycle failure classes (telemetry_unavailable, hook_protocol_error,
//!   internal_error).
//!
//! This module does **not** own:
//! * receipt emission (callers translate [`TelemetryError`] into a
//!   [`crate::LifecycleReceipt`]),
//! * the placement/payload pipeline (issue #4 owns asset rendering),
//! * adapter manifest registration (issue #6 owns the manifest registry —
//!   this module merely reports `support` states the registry can attach
//!   to a `context_pressure` claim),
//! * the hook protocol command strings (issue #3),
//! * lifecycle routing (issue #7),
//! * any client-side state, prompt semantics, or continuity vocabulary.
//!   Specifically: no memory, recall, promotion, compaction policy, radar,
//!   or governance reasoning. Lifeloop reports the *signal*; clients
//!   decide what it means.
//!
//! # CCD compatibility
//!
//! Existing `CCD_*` env vars (e.g. `CCD_CLAUDE_HOME`,
//! `CCD_CONTEXT_WINDOW_TOKENS`, `CCD_HOST_MODEL`) are honored as
//! compatibility inputs through `lifeloop.v0.x`. Each has a
//! `LIFELOOP_*` alias; when both are set, the `LIFELOOP_*` value wins
//! and a single bounded warning is recorded per resolved key per
//! process. Removal criteria are tracked in
//! `docs/tombstones/lifeloop.v0.md`.

use std::path::{Path, PathBuf};
use std::sync::Mutex;
use std::time::{SystemTime, UNIX_EPOCH};

use serde::{Deserialize, Serialize};

pub mod claude;
pub mod codex;
pub mod gemini;
pub mod host;
pub mod opencode;

pub(crate) const MAX_TELEMETRY_LOG_BYTES: u64 = 64 * 1024 * 1024;

// ============================================================================
// Neutral observation types
// ============================================================================

/// A neutral lifecycle-pressure observation extracted from harness
/// telemetry. Carries everything a `context.pressure_observed` event or a
/// receipt `telemetry_summary` needs, with no harness-private fields.
///
/// Wire-stable field names; `Option`s are `skip_serializing_if` so absent
/// signals stay absent on the wire (callers must not rely on `null`
/// vs missing for these — they are not required-nullable).
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct PressureObservation {
    /// Stable adapter id (matches `AdapterManifest::adapter_id`, e.g.
    /// `"claude"`, `"codex"`, `"gemini"`, `"opencode"`).
    pub adapter_id: String,
    /// Adapter version when discoverable from the telemetry source.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub adapter_version: Option<String>,
    /// Seconds since UNIX epoch at which this observation was sourced
    /// (typically the underlying log/session-file mtime).
    pub observed_at_epoch_s: u64,
    /// Model identifier when surfaced by the telemetry source or env.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub model_name: Option<String>,
    /// Most recent prompt-side token count (window-relative).
    #[serde(skip_serializing_if = "Option::is_none")]
    pub total_tokens: Option<u64>,
    /// Adapter-reported context window in tokens.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub context_window_tokens: Option<u64>,
    /// Percentage of the context window consumed (0..=100), when both
    /// numerator and denominator are available.
    #[serde(skip_serializing_if = "Option::is_none")]
    pub context_used_pct: Option<u8>,
    /// `Some(true)` if the adapter signaled a compaction/compression event
    /// in the observed session window. `None` means "no signal" (not
    /// "definitely false").
    #[serde(skip_serializing_if = "Option::is_none")]
    pub compaction_signal: Option<bool>,
    /// Granular usage breakdown, when the source provides it.
    #[serde(skip_serializing_if = "TokenUsage::is_empty", default)]
    pub usage: TokenUsage,
}

/// Per-side token usage breakdown. All fields default to zero so the type
/// stays format-agnostic for `Deserialize`.
#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct TokenUsage {
    #[serde(default, skip_serializing_if = "is_zero_u64")]
    pub input_tokens: u64,
    #[serde(default, skip_serializing_if = "is_zero_u64")]
    pub output_tokens: u64,
    #[serde(default, skip_serializing_if = "is_zero_u64")]
    pub cache_creation_input_tokens: u64,
    #[serde(default, skip_serializing_if = "is_zero_u64")]
    pub cache_read_input_tokens: u64,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub blended_total_tokens: Option<u64>,
}

impl TokenUsage {
    pub fn is_empty(&self) -> bool {
        self.input_tokens == 0
            && self.output_tokens == 0
            && self.cache_creation_input_tokens == 0
            && self.cache_read_input_tokens == 0
            && self.blended_total_tokens.unwrap_or(0) == 0
    }
}

fn is_zero_u64(value: &u64) -> bool {
    *value == 0
}

// ============================================================================
// Errors
// ============================================================================

/// Telemetry-side errors. The variants name the failure classes the spec
/// uses for telemetry-derived receipts:
///
/// * [`TelemetryError::Unavailable`] → `telemetry_unavailable`
/// * [`TelemetryError::HookProtocol`] → `hook_protocol_error`
/// * [`TelemetryError::Internal`] → `internal_error` (the only one of the
///   three currently named in `docs/specs/lifecycle-contract/body.md`).
///
/// The first two are *pending* as `LifecycleReceipt::failure_class`
/// variants — they are added to the receipt enum in a follow-up issue
/// once the spec body names them. Until then, callers wishing to emit a
/// failed receipt for these conditions should use `internal_error` and
/// attach the precise class via `warnings`.
#[derive(Debug)]
pub enum TelemetryError {
    /// The telemetry source (file, directory, db) was missing, empty, or
    /// stale. Distinct from `internal_error` because it's an expected,
    /// observable absence rather than a Lifeloop bug.
    Unavailable(String),
    /// The shape of a parsed telemetry artifact violated the adapter's
    /// hook protocol contract (e.g. expected JSONL but lines were not
    /// objects). Distinct from `Unavailable` because something *was*
    /// there and it was wrong.
    HookProtocol(String),
    /// Lifeloop failed unexpectedly while parsing telemetry.
    Internal(String),
}

impl TelemetryError {
    /// Stable wire string for the failure class this error maps to.
    pub fn failure_class(&self) -> &'static str {
        match self {
            Self::Unavailable(_) => "telemetry_unavailable",
            Self::HookProtocol(_) => "hook_protocol_error",
            Self::Internal(_) => "internal_error",
        }
    }
}

impl std::fmt::Display for TelemetryError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::Unavailable(msg) => write!(f, "telemetry_unavailable: {msg}"),
            Self::HookProtocol(msg) => write!(f, "hook_protocol_error: {msg}"),
            Self::Internal(msg) => write!(f, "internal_error: {msg}"),
        }
    }
}

impl std::error::Error for TelemetryError {}

impl From<std::io::Error> for TelemetryError {
    fn from(error: std::io::Error) -> Self {
        match error.kind() {
            std::io::ErrorKind::NotFound => Self::Unavailable(error.to_string()),
            _ => Self::Internal(error.to_string()),
        }
    }
}

pub type TelemetryResult<T> = Result<T, TelemetryError>;

pub(crate) fn read_file_bounded(path: &Path, label: &str) -> TelemetryResult<Vec<u8>> {
    let metadata = path.metadata().map_err(TelemetryError::from)?;
    if metadata.len() > MAX_TELEMETRY_LOG_BYTES {
        return Err(TelemetryError::Unavailable(format!(
            "{label} exceeds {MAX_TELEMETRY_LOG_BYTES} bytes: {}",
            path.display()
        )));
    }
    let file = std::fs::File::open(path).map_err(TelemetryError::from)?;
    let mut bytes = Vec::new();
    use std::io::Read;
    file.take(MAX_TELEMETRY_LOG_BYTES + 1)
        .read_to_end(&mut bytes)
        .map_err(TelemetryError::from)?;
    if bytes.len() as u64 > MAX_TELEMETRY_LOG_BYTES {
        return Err(TelemetryError::Unavailable(format!(
            "{label} exceeds {MAX_TELEMETRY_LOG_BYTES} bytes: {}",
            path.display()
        )));
    }
    Ok(bytes)
}

pub(crate) fn read_file_to_string_bounded(path: &Path, label: &str) -> TelemetryResult<String> {
    let bytes = read_file_bounded(path, label)?;
    String::from_utf8(bytes).map_err(|err| {
        TelemetryError::HookProtocol(format!("{label} is not UTF-8: {}: {err}", path.display()))
    })
}

// ============================================================================
// Env var resolution: LIFELOOP_* wins over CCD_*, with bounded warning.
// ============================================================================

/// A single CCD→Lifeloop env-var alias.
///
/// Each adapter declares the keys it cares about as a `&'static [EnvAlias]`.
/// When both the `lifeloop` and `ccd_compat` keys are set in the process
/// environment, the `lifeloop` value wins and [`EnvWarningSink`] records a
/// single bounded warning per resolved key per process.
#[derive(Debug, Clone, Copy)]
pub struct EnvAlias {
    pub lifeloop: &'static str,
    pub ccd_compat: &'static str,
}

/// Sink for env-precedence warnings.
///
/// One global instance ([`env_warning_sink`]) deduplicates warnings by the
/// `lifeloop` key name so a process that resolves the same alias many
/// times only emits one warning. Tests can inspect [`drain`] to assert
/// precedence behavior.
///
/// [`drain`]: EnvWarningSink::drain
#[derive(Debug, Default)]
pub struct EnvWarningSink {
    inner: Mutex<EnvWarningInner>,
}

#[derive(Debug, Default)]
struct EnvWarningInner {
    seen: Vec<String>,
    queued: Vec<EnvPrecedenceWarning>,
}

/// Bounded warning emitted once per resolved alias when both
/// `LIFELOOP_*` and `CCD_*` are set. The `lifeloop` value wins; the
/// `ccd_compat` value is shadowed.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct EnvPrecedenceWarning {
    pub lifeloop_key: &'static str,
    pub ccd_compat_key: &'static str,
}

impl EnvWarningSink {
    fn note(&self, alias: EnvAlias) {
        let mut inner = self.inner.lock().expect("env warning sink poisoned");
        if inner.seen.iter().any(|k| k == alias.lifeloop) {
            return;
        }
        inner.seen.push(alias.lifeloop.to_string());
        inner.queued.push(EnvPrecedenceWarning {
            lifeloop_key: alias.lifeloop,
            ccd_compat_key: alias.ccd_compat,
        });
    }

    /// Drain queued warnings (FIFO). Each warning is yielded at most
    /// once: a key that has already been drained will not appear again
    /// in a later call, even if its alias resolves repeatedly.
    pub fn drain(&self) -> Vec<EnvPrecedenceWarning> {
        let mut inner = self.inner.lock().expect("env warning sink poisoned");
        std::mem::take(&mut inner.queued)
    }

    /// Test-only: forget all dedupe state. Production code never calls
    /// this; the sink is intended to live for the process lifetime.
    #[doc(hidden)]
    pub fn reset_for_tests(&self) {
        let mut inner = self.inner.lock().expect("env warning sink poisoned");
        inner.seen.clear();
        inner.queued.clear();
    }
}

/// Process-wide warning sink used by [`resolve_env_string`] and
/// [`resolve_env_u64`]. Callers can `drain` to surface these warnings on
/// their preferred channel.
pub fn env_warning_sink() -> &'static EnvWarningSink {
    use std::sync::OnceLock;
    static SINK: OnceLock<EnvWarningSink> = OnceLock::new();
    SINK.get_or_init(EnvWarningSink::default)
}

/// Resolve an env-var string through the alias list. The first alias
/// whose `lifeloop` key is set wins; otherwise the first alias whose
/// `ccd_compat` key is set wins. Whenever an alias has *both* sides set,
/// a precedence warning is recorded (once per `lifeloop` key per
/// process) regardless of which alias actually carried the resolution.
pub fn resolve_env_string(aliases: &[EnvAlias]) -> Option<String> {
    resolve_env_string_with(aliases, &|name| std::env::var(name).ok())
}

/// Like [`resolve_env_string`] but reads through a closure (for tests
/// that don't want to mutate process env).
pub fn resolve_env_string_with(
    aliases: &[EnvAlias],
    read: &dyn Fn(&str) -> Option<String>,
) -> Option<String> {
    let mut chosen: Option<String> = None;

    for alias in aliases {
        let lifeloop_value = read(alias.lifeloop)
            .map(|v| v.trim().to_owned())
            .filter(|v| !v.is_empty());
        let ccd_value = read(alias.ccd_compat)
            .map(|v| v.trim().to_owned())
            .filter(|v| !v.is_empty());

        if lifeloop_value.is_some() && ccd_value.is_some() {
            env_warning_sink().note(*alias);
        }

        if chosen.is_none() {
            chosen = lifeloop_value.or(ccd_value);
        }
    }

    chosen
}

/// Resolve an env alias as `u64`, parsing the string form.
pub fn resolve_env_u64(aliases: &[EnvAlias]) -> Option<u64> {
    resolve_env_string(aliases).and_then(|v| v.parse().ok())
}

/// `Lifeloop`/`CCD` general context-window fallback. Adapter readers
/// should consult their adapter-specific alias first, then fall back to
/// this. The general fallback exists so users with custom or local
/// setups don't have to set a runtime-prefixed env var.
pub const GENERAL_CONTEXT_WINDOW_ALIASES: &[EnvAlias] = &[EnvAlias {
    lifeloop: "LIFELOOP_CONTEXT_WINDOW_TOKENS",
    ccd_compat: "CCD_CONTEXT_WINDOW_TOKENS",
}];

/// Generic host-model fallback used by every adapter when a more
/// specific alias does not resolve.
pub const GENERAL_HOST_MODEL_ALIASES: &[EnvAlias] = &[EnvAlias {
    lifeloop: "LIFELOOP_HOST_MODEL",
    ccd_compat: "CCD_HOST_MODEL",
}];

pub fn general_context_window() -> Option<u64> {
    resolve_env_u64(GENERAL_CONTEXT_WINDOW_ALIASES)
}

pub fn general_host_model() -> Option<String> {
    resolve_env_string(GENERAL_HOST_MODEL_ALIASES)
}

// ============================================================================
// Filesystem helpers (shared by per-adapter readers)
// ============================================================================

/// Seconds since UNIX epoch for the file's mtime, or `None` if the file
/// is missing.
pub fn file_mtime_epoch_s(path: &Path) -> TelemetryResult<Option<u64>> {
    let metadata = match std::fs::metadata(path) {
        Ok(m) => m,
        Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(None),
        Err(error) => return Err(TelemetryError::Internal(error.to_string())),
    };
    let modified = metadata
        .modified()
        .map_err(|e| TelemetryError::Internal(e.to_string()))?;
    let epoch_s = modified
        .duration_since(UNIX_EPOCH)
        .map_err(|e| TelemetryError::Internal(format!("mtime before UNIX_EPOCH: {e}")))?
        .as_secs();
    Ok(Some(epoch_s))
}

/// Default recency window (seconds) for considering a telemetry artifact
/// fresh enough to drive a `context.pressure_observed` event. Mirrors the
/// 30-minute threshold the CCD readers used.
pub const RECENT_ACTIVITY_SECS: u64 = 30 * 60;

pub fn now_epoch_s() -> TelemetryResult<u64> {
    SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map(|d| d.as_secs())
        .map_err(|e| TelemetryError::Internal(format!("system clock before UNIX_EPOCH: {e}")))
}

pub fn is_recent(epoch_s: u64) -> TelemetryResult<bool> {
    Ok(now_epoch_s()?.saturating_sub(epoch_s) <= RECENT_ACTIVITY_SECS)
}

pub fn home_dir() -> TelemetryResult<PathBuf> {
    match std::env::var_os("HOME") {
        Some(home) => Ok(PathBuf::from(home)),
        None => Err(TelemetryError::Unavailable(
            "HOME environment variable is not set".into(),
        )),
    }
}

pub fn compute_pct(total_tokens: u64, context_window: Option<u64>) -> Option<u8> {
    let cw = context_window?;
    if cw == 0 {
        return None;
    }
    Some(((total_tokens.saturating_mul(100)) / cw).min(100) as u8)
}

// ============================================================================
// JSON probing helpers (shared by per-adapter readers)
// ============================================================================

/// Recursively search a JSON value for the first matching string by
/// candidate key names. Descends into objects and arrays depth-first.
pub fn string_key(value: &serde_json::Value, keys: &[&str]) -> Option<String> {
    match value {
        serde_json::Value::Object(map) => {
            for key in keys {
                if let Some(serde_json::Value::String(found)) = map.get(*key) {
                    return Some(found.clone());
                }
            }
            for child in map.values() {
                if let Some(found) = string_key(child, keys) {
                    return Some(found);
                }
            }
            None
        }
        serde_json::Value::Array(items) => items.iter().find_map(|i| string_key(i, keys)),
        _ => None,
    }
}

/// Recursively search a JSON value for the first matching numeric value
/// by candidate key names. Accepts both JSON numbers and stringified
/// integers.
pub fn number_key(value: &serde_json::Value, keys: &[&str]) -> Option<u64> {
    match value {
        serde_json::Value::Object(map) => {
            for key in keys {
                if let Some(found) = map.get(*key)
                    && let Some(number) = as_u64(found)
                {
                    return Some(number);
                }
            }
            for child in map.values() {
                if let Some(found) = number_key(child, keys) {
                    return Some(found);
                }
            }
            None
        }
        serde_json::Value::Array(items) => items.iter().find_map(|i| number_key(i, keys)),
        _ => None,
    }
}

/// Coerce a JSON value to u64, accepting both numbers and parseable
/// strings.
pub fn as_u64(value: &serde_json::Value) -> Option<u64> {
    match value {
        serde_json::Value::Number(number) => number.as_u64(),
        serde_json::Value::String(text) => text.parse().ok(),
        _ => None,
    }
}

// ============================================================================
// Tests
// ============================================================================

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

    // Tests below that touch the global `env_warning_sink` must serialize:
    // cargo test runs unit tests in parallel by default and the sink's
    // reset/drain pair races otherwise. A process-local mutex keeps the
    // test set dependency-free.
    static ENV_SINK_LOCK: Mutex<()> = Mutex::new(());
    use serde_json::json;

    #[test]
    fn as_u64_accepts_numbers_and_strings() {
        assert_eq!(as_u64(&json!(42)), Some(42));
        assert_eq!(as_u64(&json!("1024")), Some(1024));
        assert_eq!(as_u64(&json!(-1)), None);
        assert_eq!(as_u64(&json!(1.5)), None);
        assert_eq!(as_u64(&json!("hello")), None);
    }

    #[test]
    fn string_key_descends() {
        let v = json!({"outer": {"inner": {"target": "found"}}});
        assert_eq!(string_key(&v, &["target"]), Some("found".into()));
    }

    #[test]
    fn number_key_descends() {
        let v = json!({"usage": {"prompt_tokens": 200}});
        assert_eq!(number_key(&v, &["prompt_tokens"]), Some(200));
    }

    #[test]
    fn telemetry_error_failure_classes_are_stable() {
        assert_eq!(
            TelemetryError::Unavailable("x".into()).failure_class(),
            "telemetry_unavailable"
        );
        assert_eq!(
            TelemetryError::HookProtocol("x".into()).failure_class(),
            "hook_protocol_error"
        );
        assert_eq!(
            TelemetryError::Internal("x".into()).failure_class(),
            "internal_error"
        );
    }

    #[test]
    fn bounded_file_read_rejects_oversized_logs() {
        let dir = tempfile::tempdir().expect("tempdir");
        let path = dir.path().join("huge.jsonl");
        let file = std::fs::File::create(&path).expect("create file");
        file.set_len(MAX_TELEMETRY_LOG_BYTES + 1)
            .expect("sparse file length");

        let err = read_file_bounded(&path, "test log").unwrap_err();
        assert!(matches!(err, TelemetryError::Unavailable(_)));
    }

    #[test]
    fn pressure_observation_serializes_minimally() {
        let obs = PressureObservation {
            adapter_id: "claude".into(),
            adapter_version: None,
            observed_at_epoch_s: 100,
            model_name: None,
            total_tokens: Some(500),
            context_window_tokens: Some(1000),
            context_used_pct: Some(50),
            compaction_signal: None,
            usage: TokenUsage::default(),
        };
        let json = serde_json::to_value(&obs).unwrap();
        assert_eq!(json["adapter_id"], "claude");
        assert_eq!(json["observed_at_epoch_s"], 100);
        assert_eq!(json["total_tokens"], 500);
        assert!(json.get("adapter_version").is_none());
        assert!(json.get("compaction_signal").is_none());
        assert!(json.get("usage").is_none());
    }

    #[test]
    fn resolve_env_string_with_lifeloop_winning() {
        let _g = ENV_SINK_LOCK.lock().unwrap();
        env_warning_sink().reset_for_tests();
        let aliases = &[EnvAlias {
            lifeloop: "LIFELOOP_TEST_X",
            ccd_compat: "CCD_TEST_X",
        }];
        let read = |name: &str| -> Option<String> {
            match name {
                "LIFELOOP_TEST_X" => Some("ll".into()),
                "CCD_TEST_X" => Some("ccd".into()),
                _ => None,
            }
        };
        assert_eq!(resolve_env_string_with(aliases, &read), Some("ll".into()));
        let warnings = env_warning_sink().drain();
        assert_eq!(warnings.len(), 1);
        assert_eq!(warnings[0].lifeloop_key, "LIFELOOP_TEST_X");
        assert_eq!(warnings[0].ccd_compat_key, "CCD_TEST_X");
    }

    #[test]
    fn resolve_env_string_falls_back_to_ccd() {
        let _g = ENV_SINK_LOCK.lock().unwrap();
        env_warning_sink().reset_for_tests();
        let aliases = &[EnvAlias {
            lifeloop: "LIFELOOP_TEST_Y",
            ccd_compat: "CCD_TEST_Y",
        }];
        let read = |name: &str| -> Option<String> {
            match name {
                "CCD_TEST_Y" => Some("ccd-only".into()),
                _ => None,
            }
        };
        assert_eq!(
            resolve_env_string_with(aliases, &read),
            Some("ccd-only".into())
        );
        assert!(env_warning_sink().drain().is_empty());
    }

    #[test]
    fn warning_is_bounded_to_one_per_key() {
        let _g = ENV_SINK_LOCK.lock().unwrap();
        env_warning_sink().reset_for_tests();
        let aliases = &[EnvAlias {
            lifeloop: "LIFELOOP_TEST_Z",
            ccd_compat: "CCD_TEST_Z",
        }];
        let read = |name: &str| -> Option<String> {
            match name {
                "LIFELOOP_TEST_Z" => Some("ll".into()),
                "CCD_TEST_Z" => Some("ccd".into()),
                _ => None,
            }
        };
        for _ in 0..5 {
            let _ = resolve_env_string_with(aliases, &read);
        }
        let warnings = env_warning_sink().drain();
        assert_eq!(warnings.len(), 1);
        // After drain, the seen-set still holds the key, so subsequent
        // resolutions must NOT requeue it.
        for _ in 0..3 {
            let _ = resolve_env_string_with(aliases, &read);
        }
        assert!(env_warning_sink().drain().is_empty());
    }
}