openlatch-client 0.1.18

//! Prefix-churn classification (D-12).
//!
//! Compares this request's prefix against the **previous** request in the same
//! `(install_id, agent_session)` and emits a deterministic *classification* plus
//! offsets — **never content** (F-34). The churning block content is written to
//! the local [`super::retention`] store keyed by a host-local `finding_id`, so
//! `openlatch boundary explain <finding_id>` can show it **on this host only**.
//!
//! The classification is what makes a prefix finding actionable while the prompt
//! stays local — the mechanism behind the six `prefix_*` columns (C-5) that feed
//! `/economics/prefix-findings`.

use dashmap::DashMap;

/// Cap on the prefix bytes retained per session for diffing — bounds memory
/// against a pathological large-body caller. Divergence beyond this is not
/// classified (the finding is simply `None`).
const MAX_DIFF_PREFIX: usize = 512 * 1024;

/// Cap on the number of `(install, session)` prefixes tracked at once.
const MAX_TRACKED: usize = 4096;

/// Bounded window read forward from the divergence point to extract the changed
/// token for classification + local retention.
const BLOCK_WINDOW: usize = 512;

/// Which cache layer diverged (frozen enum `churn_layer`). `tools` invalidates
/// everything downstream; `messages` the least.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum ChurnLayer {
    Tools,
    System,
    Messages,
}

impl ChurnLayer {
    pub fn as_str(&self) -> &'static str {
        match self {
            ChurnLayer::Tools => "tools",
            ChurnLayer::System => "system",
            ChurnLayer::Messages => "messages",
        }
    }
}

/// The shape of a churning block (frozen enum `churn_class`).
///
/// > **`freeform` has NO producing rule here (C-12b).** It is in the frozen enum
/// > but is never emitted — an unclassifiable block defaults to `unknown`.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum ChurnClass {
    Timestamp,
    Counter,
    PathList,
    Identifier,
    Unknown,
}

impl ChurnClass {
    pub fn as_str(&self) -> &'static str {
        match self {
            ChurnClass::Timestamp => "timestamp",
            ChurnClass::Counter => "counter",
            ChurnClass::PathList => "path_list",
            ChurnClass::Identifier => "identifier",
            ChurnClass::Unknown => "unknown",
        }
    }
}

/// A prefix-churn finding — **classification + offsets only**, no content. Lands
/// in the six `prefix_*` wire fields (C-5). The `block` is retained separately,
/// on-host, and is not part of what is emitted.
#[derive(Clone, Debug)]
pub struct ChurnFinding {
    pub divergence_offset: u64,
    pub churn_layer: ChurnLayer,
    pub churn_class: ChurnClass,
    pub churn_block_index: u64,
    pub churn_byte_len: u64,
    pub finding_id: String,
    /// The actual changed block — **host-local only, never emitted.** Written to
    /// the retention store by the caller for `boundary explain`.
    pub block: String,
}

/// Per-session previous-prefix store, shared on `BoundaryState`.
#[derive(Default)]
pub struct ChurnTracker {
    prev: DashMap<String, Vec<u8>>,
}

impl ChurnTracker {
    /// Compare `cur_body` against the previous body for this session, then record
    /// `cur_body` as the new previous. Returns a finding when the prefix diverged
    /// (`None` on the first request of a session or when identical).
    pub fn observe(&self, install: &str, session: &str, cur_body: &[u8]) -> Option<ChurnFinding> {
        let key = format!("{install}\u{1}{session}");
        let cur_prefix: Vec<u8> = cur_body[..cur_body.len().min(MAX_DIFF_PREFIX)].to_vec();

        let finding = self
            .prev
            .get(&key)
            .and_then(|prev| classify_churn(prev.value(), &cur_prefix));

        // Bound the tracked-key set; a brand-new key past the cap is not stored
        // (its next request just won't diff — safe degradation).
        if self.prev.contains_key(&key) || self.prev.len() < MAX_TRACKED {
            self.prev.insert(key, cur_prefix);
        }
        finding
    }
}

/// Core deterministic classification. `prev`/`cur` are the two prefixes.
pub fn classify_churn(prev: &[u8], cur: &[u8]) -> Option<ChurnFinding> {
    let off = first_divergence_offset(prev, cur)?;
    let layer = layer_of(cur, off);
    let (block, byte_len) = extract_block(cur, off);
    let class = classify(&block);
    let block_index = block_index(cur, off, layer);

    Some(ChurnFinding {
        divergence_offset: off as u64,
        churn_layer: layer,
        churn_class: class,
        churn_block_index: block_index,
        churn_byte_len: byte_len as u64,
        finding_id: new_finding_id(),
        block,
    })
}

/// First byte offset where the two prefixes differ. `None` when identical.
/// When one is a strict prefix of the other, they diverge at the shorter length.
fn first_divergence_offset(prev: &[u8], cur: &[u8]) -> Option<usize> {
    let min = prev.len().min(cur.len());
    for i in 0..min {
        if prev[i] != cur[i] {
            return Some(i);
        }
    }
    if prev.len() == cur.len() {
        None
    } else {
        Some(min)
    }
}

/// Which layer the divergence offset falls in, by locating the latest top-level
/// layer key at or before `off`. Defaults to `messages` (the least-invalidating
/// layer) when no key precedes the offset.
fn layer_of(cur: &[u8], off: usize) -> ChurnLayer {
    let head = &cur[..off.min(cur.len())];
    // The layer whose key appears LATEST before the divergence owns it; default
    // to the least-invalidating `messages` when no key precedes the offset. Two
    // distinct keys can never start at the same byte offset, so there is never a
    // tie for `max_by_key` to resolve.
    let candidates: [(&[u8], ChurnLayer); 3] = [
        (b"\"tools\"", ChurnLayer::Tools),
        (b"\"system\"", ChurnLayer::System),
        (b"\"messages\"", ChurnLayer::Messages),
    ];
    candidates
        .into_iter()
        .filter_map(|(key, layer)| rfind(head, key).map(|p| (p, layer)))
        .max_by_key(|&(p, _)| p)
        .map(|(_, layer)| layer)
        .unwrap_or(ChurnLayer::Messages)
}

/// Extract the whole changed **value** around the divergence, bounded to
/// [`BLOCK_WINDOW`] each way.
///
/// The divergence byte usually lands *inside* a value (e.g. the last digit of a
/// date), so we back up to the value's opening delimiter and read forward to its
/// close — classifying `2026-07-23` as a whole, not just the trailing `3`.
fn extract_block(cur: &[u8], off: usize) -> (String, usize) {
    let off = off.min(cur.len());

    // Back up to the byte after the last opening delimiter before `off`. `/` is
    // deliberately NOT a delimiter, so a path value stays whole.
    let back_limit = off.saturating_sub(BLOCK_WINDOW);
    let mut start = off;
    while start > back_limit {
        if matches!(
            cur[start - 1],
            b'"' | b':' | b',' | b'[' | b'{' | b' ' | b'\n' | b'\t'
        ) {
            break;
        }
        start -= 1;
    }

    // Read forward to the next closing/structural delimiter.
    let fwd_limit = (off + BLOCK_WINDOW).min(cur.len());
    let mut end = off;
    while end < fwd_limit {
        if matches!(cur[end], b'"' | b',' | b'}' | b']' | b'\n') {
            break;
        }
        end += 1;
    }
    if end <= start {
        end = (start + 1).min(cur.len());
    }

    let block = &cur[start..end];
    (String::from_utf8_lossy(block).to_string(), block.len())
}

/// Approximate block index within the layer: count object-opens before the
/// divergence within the layer region. Lets a developer find the block locally.
fn block_index(cur: &[u8], off: usize, layer: ChurnLayer) -> u64 {
    let key: &[u8] = match layer {
        ChurnLayer::Tools => b"\"tools\"",
        ChurnLayer::System => b"\"system\"",
        ChurnLayer::Messages => b"\"messages\"",
    };
    let head = &cur[..off.min(cur.len())];
    let layer_start = rfind(head, key).unwrap_or(0);
    head[layer_start..].iter().filter(|&&b| b == b'{').count() as u64
}

/// Deterministic pattern classification of a changed token. Never returns
/// `freeform` (C-12b) — an unclassifiable token is `unknown`.
fn classify(raw: &str) -> ChurnClass {
    let t = raw.trim().trim_matches('"').trim();
    if t.is_empty() {
        return ChurnClass::Unknown;
    }
    if looks_like_timestamp(t) {
        return ChurnClass::Timestamp;
    }
    if t.bytes().all(|b| b.is_ascii_digit()) {
        return ChurnClass::Counter;
    }
    if looks_like_path(t) {
        return ChurnClass::PathList;
    }
    if looks_like_identifier(t) {
        return ChurnClass::Identifier;
    }
    ChurnClass::Unknown
}

/// ISO-8601-ish: `YYYY-MM-DD` optionally followed by `T…`, or an `HH:MM:SS` time.
fn looks_like_timestamp(t: &str) -> bool {
    let b = t.as_bytes();
    // YYYY-MM-DD prefix.
    if b.len() >= 10
        && b[0..4].iter().all(u8::is_ascii_digit)
        && b[4] == b'-'
        && b[5..7].iter().all(u8::is_ascii_digit)
        && b[7] == b'-'
        && b[8..10].iter().all(u8::is_ascii_digit)
    {
        return true;
    }
    // HH:MM:SS time.
    if b.len() >= 8
        && b[0..2].iter().all(u8::is_ascii_digit)
        && b[2] == b':'
        && b[3..5].iter().all(u8::is_ascii_digit)
        && b[5] == b':'
        && b[6..8].iter().all(u8::is_ascii_digit)
    {
        return true;
    }
    false
}

/// Filesystem-path shape: starts with `/`, `./`, `~/`, a Windows drive, or holds
/// two or more path separators.
fn looks_like_path(t: &str) -> bool {
    t.starts_with('/')
        || t.starts_with("./")
        || t.starts_with("../")
        || t.starts_with("~/") // portability-ok: shape detector over agent text, resolves nothing
        || (t.len() > 2 && t.as_bytes()[1] == b':' && (t.contains('\\') || t.contains('/')))
        || t.matches('/').count() >= 2
}

/// UUID shape (8-4-4-4-12 hex) or a long hex/opaque token run.
fn looks_like_identifier(t: &str) -> bool {
    if is_uuid(t) {
        return true;
    }
    // A long, single-token hex/base-ish run with no spaces reads as an id/hash.
    t.len() >= 16
        && !t.contains(' ')
        && t.bytes()
            .all(|b| b.is_ascii_alphanumeric() || b == b'_' || b == b'-')
        && t.bytes().any(|b| b.is_ascii_digit())
}

fn is_uuid(t: &str) -> bool {
    let parts: Vec<&str> = t.split('-').collect();
    parts.len() == 5
        && [8, 4, 4, 4, 12]
            == [
                parts[0].len(),
                parts[1].len(),
                parts[2].len(),
                parts[3].len(),
                parts[4].len(),
            ]
        && parts
            .iter()
            .all(|p| p.bytes().all(|b| b.is_ascii_hexdigit()))
}

/// Last occurrence of `needle` in `haystack`.
fn rfind(haystack: &[u8], needle: &[u8]) -> Option<usize> {
    if needle.is_empty() || haystack.len() < needle.len() {
        return None;
    }
    (0..=haystack.len() - needle.len())
        .rev()
        .find(|&i| &haystack[i..i + needle.len()] == needle)
}

/// A host-local finding handle. Resolves only on this machine via retention.
fn new_finding_id() -> String {
    // UUIDv7 keeps the id time-sortable; the `fnd_` prefix marks its namespace.
    format!("fnd_{}", uuid::Uuid::now_v7())
}

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

    #[test]
    fn identical_prefix_yields_no_finding() {
        let body = br#"{"messages":[{"role":"user","content":"hi"}]}"#;
        assert!(classify_churn(body, body).is_none());
    }

    #[test]
    fn timestamp_change_is_classified_timestamp_no_content() {
        let prev = br#"{"system":[{"text":"now is 2026-07-22"}],"messages":[]}"#;
        let cur = br#"{"system":[{"text":"now is 2026-07-23"}],"messages":[]}"#;
        let f = classify_churn(prev, cur).expect("divergence must be found");
        assert_eq!(f.churn_class, ChurnClass::Timestamp);
        assert_eq!(f.churn_layer, ChurnLayer::System);
        // The finding carries a class + offsets, and its id is host-local.
        assert!(f.finding_id.starts_with("fnd_"));
        assert!(f.divergence_offset > 0);
    }

    #[test]
    fn counter_and_path_and_identifier() {
        let a = br#"{"messages":[{"n":41}]}"#;
        let b = br#"{"messages":[{"n":42}]}"#;
        assert_eq!(
            classify_churn(a, b).unwrap().churn_class,
            ChurnClass::Counter
        );

        let p1 = br#"{"messages":[{"cwd":"/home/a/one"}]}"#;
        let p2 = br#"{"messages":[{"cwd":"/home/a/two/three"}]}"#;
        assert_eq!(
            classify_churn(p1, p2).unwrap().churn_class,
            ChurnClass::PathList
        );

        let i1 = br#"{"messages":[{"id":"550e8400-e29b-41d4-a716-446655440000"}]}"#;
        let i2 = br#"{"messages":[{"id":"550e8400-e29b-41d4-a716-4466554400ff"}]}"#;
        assert_eq!(
            classify_churn(i1, i2).unwrap().churn_class,
            ChurnClass::Identifier
        );
    }

    #[test]
    fn never_emits_freeform() {
        // Whatever the block, the class is one of the FIVE producible values,
        // never `freeform` (C-12b).
        let prev = br#"{"messages":[{"x":"lorem ipsum dolor"}]}"#;
        let cur = br#"{"messages":[{"x":"a wholly different sentence here"}]}"#;
        let f = classify_churn(prev, cur).unwrap();
        assert_ne!(f.churn_class.as_str(), "freeform");
        assert_eq!(f.churn_class, ChurnClass::Unknown);
    }

    #[test]
    fn tracker_first_request_has_no_finding_then_diffs() {
        let t = ChurnTracker::default();
        let r1 = br#"{"messages":[{"n":1}]}"#;
        assert!(t.observe("agt_1", "sess", r1).is_none(), "first request");
        let r2 = br#"{"messages":[{"n":2}]}"#;
        let f = t.observe("agt_1", "sess", r2).expect("second diverges");
        assert_eq!(f.churn_class, ChurnClass::Counter);
    }

    #[test]
    fn layer_defaults_to_messages() {
        // Divergence before any layer key → default messages.
        let a = br#"{"model":"claude-opus-4-8","messages":[]}"#;
        let b = br#"{"model":"claude-sonnet-5x","messages":[]}"#;
        let f = classify_churn(a, b).unwrap();
        assert_eq!(f.churn_layer, ChurnLayer::Messages);
    }
}