openlatch-client 0.3.3

//! Parsed-structure view of the request prefix, across requests in one session
//! (Model Boundary **D-28**).
//!
//! Acting on L-0 needs two facts the byte-level churn detector
//! ([`super::churn`]) structurally cannot give:
//!
//! 1. **`H` — the measured horizon.** How many consecutive requests in this
//!    session have carried a byte-identical cacheable prefix. L-0's net is not
//!    per-turn (PRD → "`tokens_net` — the derivation" → L-0): a breakpoint buys
//!    future cache *reads* in exchange for one *write*, so the decision only
//!    exists over turns. `H` is **measured, never forecast** — it is repetition
//!    we have already been billed for, which is also what keeps the decision
//!    replayable (D-05) with no probability model and no model call (D-04).
//! 2. **Per-block stability.** Which `system[]` block is the volatile one, and
//!    which blocks behind it have never changed.
//!
//! ## Why not reuse `churn.rs`
//!
//! `churn` answers a different question well and this one not at all. It is
//! **byte-level and approximate by design**: its `churn_block_index` counts `{`
//! bytes, which any nested object inflates; it holds exactly **one** previous
//! prefix per session with no per-block history; and it reports the *first*
//! divergence offset rather than a per-position change record. That is the
//! right shape for a human-facing finding ("something around here churns") and
//! the wrong shape for a transform that must know precisely which block to move
//! and how much stable content sits behind it. Reordering off an inflated index
//! would move the wrong block.
//!
//! The two coexist deliberately: `churn` keeps observing the **original**
//! bytes, so prefix findings continue to describe what the *agent* does rather
//! than what we did to its request.
//!
//! ## Bounds
//!
//! Everything here is bounded and cheap: one hash per tracked block per
//! request, [`MAX_TRACKED_SESSIONS`] sessions, [`MAX_TRACKED_BLOCKS`] blocks per
//! layer. Past either cap the session is simply not tracked, and L-0 declines
//! rather than acting on a partial view.

use dashmap::DashMap;
use serde_json::Value;

/// Cap on tracked `(install, session)` pairs. Mirrors `churn::MAX_TRACKED`.
const MAX_TRACKED_SESSIONS: usize = 4096;

/// Cap on `system[]` blocks whose stability is tracked. A prompt with more
/// blocks than this is not a shape L-0 reasons about; it declines.
pub const MAX_TRACKED_BLOCKS: usize = 64;

/// The 64-bit FNV-1a offset basis / prime. A non-cryptographic hash is correct
/// here: the question is "are these the same bytes", and a collision costs one
/// missed or one spurious volatility observation, never a wrong forward.
const FNV_OFFSET: u64 = 0xcbf2_9ce4_8422_2325;
const FNV_PRIME: u64 = 0x0000_0100_0000_01b3;

/// What one observation tells the acting engine about this session's prefix.
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct PrefixShape {
    /// `H` — consecutive requests, **this one included**, whose whole cacheable
    /// prefix (`tools` + `system`) hashed identically. A first request is `1`,
    /// which is why a first request never fires `insert_breakpoints`: a session
    /// with one turn has nothing to amortise a write over.
    pub horizon: u32,
    /// How many requests with a `system[]` array have been observed in this
    /// session, this one included. Stability claims are meaningless below 2.
    pub observations: u32,
    /// Per-`system[]` index: did this block's bytes change on **this** request?
    ///
    /// The "landed on a churn turn" test. A reorder is only free on a turn where
    /// the moved block changed anyway — on a quiet turn it destroys a cache
    /// entry that was about to hit.
    pub changed_now: Vec<bool>,
    /// Per-`system[]` index: how many times this block's bytes have changed
    /// across the session. `0` over `observations >= 2` is the stability claim.
    pub changes: Vec<u32>,
}

impl PrefixShape {
    /// Is the `system[]` block at `index` stable — observed at least twice and
    /// never changed?
    ///
    /// Requires a second observation on purpose: after one request every block
    /// has changed zero times, which is not evidence of stability, only absence
    /// of evidence.
    pub fn is_stable(&self, index: usize) -> bool {
        self.observations >= 2 && self.changes.get(index).copied() == Some(0)
    }

    /// Is the `system[]` block at `index` genuinely volatile **and** changing on
    /// this very request?
    ///
    /// `min_changes` is the evidence bar: `2` means the block has churned at
    /// least twice, so a single hand edit to a prompt is not mistaken for a
    /// per-request timestamp.
    pub fn is_volatile_now(&self, index: usize, min_changes: u32) -> bool {
        self.changed_now.get(index).copied() == Some(true)
            && self
                .changes
                .get(index)
                .copied()
                .is_some_and(|c| c >= min_changes)
    }
}

/// Per-session record. One entry per `(install, session)`.
#[derive(Clone, Debug)]
struct SessionState {
    /// Hash of the whole cacheable prefix (`tools` + `system`) last seen.
    prefix_hash: u64,
    /// Consecutive observations with that hash, this one included.
    prefix_repeats: u32,
    /// Per-index hash of `system[]` blocks last seen.
    system_hashes: Vec<u64>,
    /// Per-index change counts.
    system_changes: Vec<u32>,
    /// Observations carrying a `system[]` array.
    observations: u32,
}

/// Per-`(install, session)` prefix-shape store, shared on `BoundaryState`.
#[derive(Default)]
pub struct PrefixTracker {
    sessions: DashMap<String, SessionState>,
}

impl PrefixTracker {
    /// Fold this request into the session's shape and return what it now knows.
    ///
    /// Synchronous, allocation-bounded, and **network-free** — it composes with
    /// the caller's `catch_unwind` exactly as `churn.observe` does.
    ///
    /// A body whose `system` is not an array (absent, or a plain string) still
    /// updates the horizon — the prefix can be stable without being blocked —
    /// but contributes no per-block record, so every stability question about it
    /// answers `false`.
    pub fn observe(&self, install: &str, session: &str, body: &Value) -> PrefixShape {
        let key = format!("{install}\u{1}{session}");
        let prefix_hash = hash_prefix(body);
        let system_hashes = system_block_hashes(body);

        // A brand-new key past the cap is not stored; its shape is the
        // first-observation shape every time, so L-0 never fires for it. Safe
        // degradation, same posture as `churn`'s bound.
        let trackable =
            self.sessions.contains_key(&key) || self.sessions.len() < MAX_TRACKED_SESSIONS;

        let mut state = self
            .sessions
            .get(&key)
            .map(|e| e.value().clone())
            .unwrap_or(SessionState {
                prefix_hash,
                prefix_repeats: 0,
                system_hashes: Vec::new(),
                system_changes: Vec::new(),
                observations: 0,
            });

        // --- horizon ---------------------------------------------------------
        state.prefix_repeats = if state.observations > 0 && state.prefix_hash == prefix_hash {
            state.prefix_repeats.saturating_add(1)
        } else {
            1
        };
        state.prefix_hash = prefix_hash;

        // --- per-block stability --------------------------------------------
        let mut changed_now = vec![false; system_hashes.len()];
        if state.observations > 0 && state.system_hashes.len() == system_hashes.len() {
            // Same arity: index-to-index comparison is meaningful.
            for (i, h) in system_hashes.iter().enumerate() {
                if state.system_hashes[i] != *h {
                    changed_now[i] = true;
                    if let Some(c) = state.system_changes.get_mut(i) {
                        *c = c.saturating_add(1);
                    }
                }
            }
        } else if state.observations > 0 {
            // The block COUNT changed, so index i is not the same block as
            // index i was. Every per-index claim is void: reset the history
            // rather than carry a comparison across a re-shaped array, which is
            // how a reorder would come to move the wrong block.
            state.system_changes = vec![0; system_hashes.len()];
            state.observations = 0;
        }
        if state.system_changes.len() != system_hashes.len() {
            state.system_changes = vec![0; system_hashes.len()];
        }
        state.system_hashes = system_hashes;
        state.observations = state.observations.saturating_add(1);

        let shape = PrefixShape {
            horizon: state.prefix_repeats,
            observations: state.observations,
            changed_now,
            changes: state.system_changes.clone(),
        };
        if trackable {
            self.sessions.insert(key, state);
        }
        shape
    }
}

/// Hash of the cacheable prefix — `tools` then `system`, the two layers that
/// render before `messages`. Deliberately excludes `messages`, which changes
/// every turn by definition and would drive the horizon to 1 forever.
fn hash_prefix(body: &Value) -> u64 {
    let mut h = FNV_OFFSET;
    for key in ["tools", "system"] {
        h = fnv(h, key.as_bytes());
        if let Some(v) = body.get(key) {
            h = fnv(h, v.to_string().as_bytes());
        }
    }
    h
}

/// Per-index hashes of the `system[]` content blocks. Empty when `system` is
/// absent, a plain string, or longer than [`MAX_TRACKED_BLOCKS`].
fn system_block_hashes(body: &Value) -> Vec<u64> {
    let Some(blocks) = body.get("system").and_then(Value::as_array) else {
        return Vec::new();
    };
    if blocks.len() > MAX_TRACKED_BLOCKS {
        return Vec::new();
    }
    blocks
        .iter()
        .map(|b| fnv(FNV_OFFSET, b.to_string().as_bytes()))
        .collect()
}

/// FNV-1a over `bytes`, continuing from `state`.
fn fnv(mut state: u64, bytes: &[u8]) -> u64 {
    for &b in bytes {
        state ^= u64::from(b);
        state = state.wrapping_mul(FNV_PRIME);
    }
    state
}

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

    fn sys(blocks: Vec<Value>) -> Value {
        json!({ "model": "claude-opus-4-8", "system": blocks, "messages": [] })
    }

    fn text(t: &str) -> Value {
        json!({ "type": "text", "text": t })
    }

    #[test]
    fn a_first_request_has_horizon_one() {
        let t = PrefixTracker::default();
        let shape = t.observe("i", "s", &sys(vec![text("a")]));
        assert_eq!(shape.horizon, 1, "one turn cannot amortise a write");
        assert_eq!(shape.observations, 1);
        assert!(
            !shape.is_stable(0),
            "one observation is not evidence of stability"
        );
    }

    #[test]
    fn an_unchanged_prefix_grows_the_horizon() {
        let t = PrefixTracker::default();
        let body = sys(vec![text("a"), text("b")]);
        assert_eq!(t.observe("i", "s", &body).horizon, 1);
        assert_eq!(t.observe("i", "s", &body).horizon, 2);
        assert_eq!(t.observe("i", "s", &body).horizon, 3);
        let shape = t.observe("i", "s", &body);
        assert_eq!(shape.horizon, 4);
        assert!(shape.is_stable(0) && shape.is_stable(1));
    }

    #[test]
    fn a_changed_prefix_resets_the_horizon_to_one() {
        let t = PrefixTracker::default();
        t.observe("i", "s", &sys(vec![text("a")]));
        t.observe("i", "s", &sys(vec![text("a")]));
        let shape = t.observe("i", "s", &sys(vec![text("CHANGED")]));
        assert_eq!(
            shape.horizon, 1,
            "the prefix we would cache is not the one we saw"
        );
    }

    #[test]
    fn messages_do_not_touch_the_horizon() {
        // `messages` change every turn by definition; counting them would pin the
        // horizon at 1 forever and the lever could never fire.
        let t = PrefixTracker::default();
        let mut body = sys(vec![text("stable")]);
        for i in 0..4 {
            body["messages"] = json!([{ "role": "user", "content": format!("turn {i}") }]);
            let shape = t.observe("i", "s", &body);
            assert_eq!(shape.horizon, u32::try_from(i).unwrap() + 1);
        }
    }

    #[test]
    fn per_block_volatility_is_located_precisely() {
        let t = PrefixTracker::default();
        for i in 0..4 {
            let body = sys(vec![
                text("frozen preamble"),
                text(&format!("current time: {i}")),
                text("frozen tail"),
            ]);
            t.observe("i", "s", &body);
        }
        let shape = t.observe(
            "i",
            "s",
            &sys(vec![
                text("frozen preamble"),
                text("current time: 99"),
                text("frozen tail"),
            ]),
        );
        assert!(shape.is_stable(0), "block 0 never changed");
        assert!(!shape.is_stable(1), "block 1 is the volatile one");
        assert!(shape.is_stable(2), "block 2 never changed");
        assert!(
            shape.is_volatile_now(1, 2),
            "and it is changing on this very request"
        );
        assert!(!shape.is_volatile_now(0, 2));
    }

    #[test]
    fn a_block_that_changed_once_is_not_yet_volatile_at_a_bar_of_two() {
        // One hand edit to a prompt must not read as a per-request timestamp.
        let t = PrefixTracker::default();
        t.observe("i", "s", &sys(vec![text("v1")]));
        let shape = t.observe("i", "s", &sys(vec![text("v2")]));
        assert!(shape.changed_now[0]);
        assert!(!shape.is_volatile_now(0, 2), "one change is not a pattern");
    }

    #[test]
    fn a_block_count_change_voids_every_per_index_claim() {
        // Index i is no longer the same block as index i was, so carrying the
        // history across would be how a reorder moves the wrong block.
        let t = PrefixTracker::default();
        for _ in 0..5 {
            t.observe("i", "s", &sys(vec![text("a"), text("b")]));
        }
        let shape = t.observe("i", "s", &sys(vec![text("a"), text("b"), text("c")]));
        assert_eq!(shape.observations, 1, "the per-index record restarted");
        assert!(
            !shape.is_stable(0),
            "no stability claim survives a re-shape"
        );
    }

    #[test]
    fn sessions_are_isolated_from_each_other() {
        let t = PrefixTracker::default();
        let body = sys(vec![text("a")]);
        t.observe("i", "s1", &body);
        t.observe("i", "s1", &body);
        assert_eq!(t.observe("i", "s2", &body).horizon, 1);
        assert_eq!(t.observe("i", "s1", &body).horizon, 3);
    }

    #[test]
    fn a_string_system_yields_no_block_record_but_still_tracks_the_horizon() {
        let t = PrefixTracker::default();
        let body = json!({ "system": "a plain string prompt", "messages": [] });
        t.observe("i", "s", &body);
        let shape = t.observe("i", "s", &body);
        assert_eq!(shape.horizon, 2);
        assert!(shape.changes.is_empty());
        assert!(!shape.is_stable(0));
    }

    #[test]
    fn observation_is_deterministic() {
        // Same session history → same shape, every time (D-05).
        let body = sys(vec![text("a"), text("b")]);
        let run = || {
            let t = PrefixTracker::default();
            t.observe("i", "s", &body);
            t.observe("i", "s", &body);
            t.observe("i", "s", &body)
        };
        let first = run();
        for _ in 0..25 {
            assert_eq!(run(), first);
        }
    }

    #[test]
    fn an_over_wide_system_array_is_not_tracked() {
        let t = PrefixTracker::default();
        let blocks: Vec<Value> = (0..MAX_TRACKED_BLOCKS + 1)
            .map(|i| text(&i.to_string()))
            .collect();
        let shape = t.observe("i", "s", &sys(blocks));
        assert!(
            shape.changes.is_empty(),
            "L-0 declines rather than acting on a partial view"
        );
    }
}