openlatch-client 0.6.3

/// In-memory duplicate event detection store with a 100ms TTL.
///
/// Deduplicates events by hashing (session_id + tool_name + canonical JSON of tool_input).
/// Duplicate events within the TTL window return the same verdict but are NOT logged,
/// preventing log spam from repeated hook calls within a single agent operation.
///
/// # Architecture
///
/// Uses `DashMap` for lock-free concurrent access. The hash key is a `u64` computed by
/// SipHash (via `DefaultHasher`). SipHash is adequate here — this is an internal dedup
/// mechanism, not a security hash. Collisions within a 100ms window are acceptable:
/// a false-positive dedup causes a single event to be skipped, not a security failure.
///
/// # Performance
///
/// PERFORMANCE: `DashMap` provides O(1) average-case reads/writes with no global lock.
/// The 100ms TTL window is small enough that the map stays tiny in practice.
use dashmap::DashMap;
use std::collections::hash_map::DefaultHasher;
use std::hash::{Hash, Hasher};
use std::time::{Duration, Instant};

/// Deduplication TTL window per EVNT-06.
const DEDUP_TTL: Duration = Duration::from_millis(100);

/// How long one tool call id stays claimed ([`DedupStore::check_and_insert_call`]).
///
/// Long because the gap it spans is not a retry burst: Cline's plugin lane and
/// its `PreToolUse` file lane post the same call, and one of them may wait out a
/// hold before the other fires. An id names exactly one call, so a long window
/// cannot fold two different ones — which is what keeps a 100 ms window right
/// for the content key and wrong for this one.
const CALL_ID_TTL: Duration = Duration::from_secs(600);

/// In-memory store for deduplicating events within a configurable TTL window.
///
/// Thread-safe via `DashMap`. All methods take `&self` (no exclusive lock needed).
pub struct DedupStore {
    /// Key → the instant it stops being a duplicate. A deadline rather than an
    /// insertion time, so keys with different windows share one map and one
    /// eviction pass.
    inner: DashMap<u64, Instant>,
    ttl: Duration,
}

impl DedupStore {
    /// Create a new `DedupStore` with the default 100ms TTL.
    pub fn new() -> Self {
        Self {
            inner: DashMap::new(),
            ttl: DEDUP_TTL,
        }
    }

    /// Check if this event is a duplicate and insert it if it is not.
    ///
    /// Returns `true` if the event was seen within the TTL window (it IS a duplicate).
    /// Returns `false` if the event is new and inserts it for future dedup checks.
    ///
    /// The dedup key is: SHA-like hash of `(session_id, tool_name, canonical_json(tool_input))`.
    pub fn check_and_insert(
        &self,
        session_id: &str,
        tool_name: &str,
        tool_input: &serde_json::Value,
    ) -> bool {
        let key = self.compute_hash(session_id, tool_name, tool_input);
        self.claim(key, self.ttl)
    }

    /// Whether this tool call id was already seen for this session and event
    /// type, claiming it for [`CALL_ID_TTL`] if not.
    ///
    /// Only for an id the AGENT sent: a derived one names a (tool, input) pair,
    /// and two separate identical calls share it.
    pub fn check_and_insert_call(&self, session_id: &str, event_type: &str, call_id: &str) -> bool {
        let key = self.compute_hash(
            session_id,
            event_type,
            &serde_json::json!({ "tool_use_id": call_id }),
        );
        self.claim(key, CALL_ID_TTL)
    }

    fn claim(&self, key: u64, ttl: Duration) -> bool {
        let now = Instant::now();
        if let Some(deadline) = self.inner.get(&key) {
            if now < *deadline.value() {
                return true; // duplicate within its window
            }
        }
        self.inner.insert(key, now + ttl);
        false
    }

    /// Compute a dedup key by hashing the event's identity fields.
    ///
    /// Uses `DefaultHasher` (SipHash) — fast and sufficient for collision avoidance
    /// within a 100ms window. This is NOT a security hash.
    ///
    /// Tool input is hashed directly from the JSON Value tree with keys sorted at each
    /// level, ensuring deterministic hashing regardless of JSON key order in the payload.
    /// This avoids the allocations of serializing to a canonical JSON string.
    fn compute_hash(
        &self,
        session_id: &str,
        tool_name: &str,
        tool_input: &serde_json::Value,
    ) -> u64 {
        let mut hasher = DefaultHasher::new();
        session_id.hash(&mut hasher);
        tool_name.hash(&mut hasher);
        hash_value(tool_input, &mut hasher);
        hasher.finish()
    }

    /// Evict all entries older than the TTL.
    ///
    /// Call periodically to prevent unbounded memory growth in long-running sessions.
    /// Under normal load (100ms TTL, <1000 events/session) the map stays very small
    /// and eviction is low-priority.
    pub fn evict_expired(&self) {
        let now = Instant::now();
        self.inner.retain(|_, deadline| now < *deadline);
    }
}

impl Default for DedupStore {
    fn default() -> Self {
        Self::new()
    }
}

/// How long a plugin-lane claim waits for its file-lane twin ([`PluginClaims`]). Longer than the
/// file lane's 3 s grace, so a plugin that answered at all has claimed before the file lane
/// checks; the margin absorbs a slow machine.
const PLUGIN_CLAIM_TTL: Duration = Duration::from_secs(10);

/// Expired claims are pruned at most this often.
const PLUGIN_CLAIMS_PRUNE_EVERY: Duration = Duration::from_secs(1);

/// Past this many keys, expired claims are pruned on every insert, whatever the interval.
const PLUGIN_CLAIMS_PRUNE_ABOVE: usize = 4096;

/// `(session, canonical JSON of the dedup identity)`.
type ClaimKey = (String, String);

/// Content claims the Cline plugin lane leaves for the file lane's copy of the same action.
///
/// A host that loads the plugin reports each tool call and each prompt through both lanes, and a
/// file-lane tool event on VS Code carries no call id to fold it by. So the plugin's report claims
/// its CONTENT, and the file lane's deferred copy consumes one claim: a hit is folded into the
/// plugin's event, a miss is stored.
///
/// A claim is a COUNT per key, not a presence flag: two identical calls fold one-for-one, and a
/// repeated call whose plugin event is missing keeps its file event.
#[derive(Default)]
pub struct PluginClaims {
    inner: DashMap<ClaimKey, (u32, Instant)>,
    last_prune: std::sync::Mutex<Option<Instant>>,
}

impl PluginClaims {
    fn key(session_id: &str, identity: &serde_json::Value) -> ClaimKey {
        let canonical = serde_json_canonicalizer::to_string(identity).unwrap_or_default();
        (session_id.to_string(), canonical)
    }

    /// Drops expired claims, at most once per [`PLUGIN_CLAIMS_PRUNE_EVERY`] unless the map is
    /// past the hard cap — so claims no file lane ever consumes (the CLI has none) do not pile
    /// up.
    fn prune(&self, now: Instant) {
        {
            let mut last = self.last_prune.lock().unwrap_or_else(|e| e.into_inner());
            let due = last.is_none_or(|at| now.duration_since(at) >= PLUGIN_CLAIMS_PRUNE_EVERY);
            if !due && self.inner.len() <= PLUGIN_CLAIMS_PRUNE_ABOVE {
                return;
            }
            *last = Some(now);
        }
        self.inner.retain(|_, (_, deadline)| now < *deadline);
    }

    /// Adds one claim for `(session_id, identity)`.
    pub fn insert(&self, session_id: &str, identity: &serde_json::Value) {
        let now = Instant::now();
        self.prune(now);
        let mut entry = self
            .inner
            .entry(Self::key(session_id, identity))
            .or_insert((0, now));
        let (count, deadline) = entry.value_mut();
        if now >= *deadline {
            *count = 0;
        }
        *count += 1;
        *deadline = now + PLUGIN_CLAIM_TTL;
    }

    /// Consumes one live claim for `(session_id, identity)`; `true` when there was one.
    pub fn consume(&self, session_id: &str, identity: &serde_json::Value) -> bool {
        let now = Instant::now();
        let key = Self::key(session_id, identity);
        let Some(mut entry) = self.inner.get_mut(&key) else {
            return false;
        };
        let (count, deadline) = entry.value_mut();
        let hit = now < *deadline && *count > 0;
        if hit {
            *count -= 1;
        }
        let spent = *count == 0 || now >= *deadline;
        drop(entry);
        if spent {
            self.inner
                .remove_if(&key, |_, (count, deadline)| *count == 0 || now >= *deadline);
        }
        hit
    }

    #[cfg(test)]
    fn len(&self) -> usize {
        self.inner.len()
    }
}

/// Hash a JSON value directly into the hasher with keys sorted at each nesting level.
///
/// This avoids allocating intermediate strings or BTreeMaps — the value tree is walked
/// once, feeding bytes directly into the hasher. Type discriminant tags ensure that
/// `"42"` (string) and `42` (number) produce different hashes.
///
/// Defensive measure: object keys are collected into a Vec and sorted before hashing,
/// ensuring deterministic output even if `preserve_order` is enabled by a transitive dep.
fn hash_value(value: &serde_json::Value, hasher: &mut impl Hasher) {
    match value {
        serde_json::Value::Null => 0u8.hash(hasher),
        serde_json::Value::Bool(b) => {
            1u8.hash(hasher);
            b.hash(hasher);
        }
        serde_json::Value::Number(n) => {
            2u8.hash(hasher);
            // Hash the debug representation — covers i64, u64, f64 uniformly
            format!("{n}").hash(hasher);
        }
        serde_json::Value::String(s) => {
            3u8.hash(hasher);
            s.hash(hasher);
        }
        serde_json::Value::Array(arr) => {
            4u8.hash(hasher);
            arr.len().hash(hasher);
            for item in arr {
                hash_value(item, hasher);
            }
        }
        serde_json::Value::Object(map) => {
            5u8.hash(hasher);
            map.len().hash(hasher);
            // Sort keys for deterministic hashing regardless of insertion order
            let mut keys: Vec<&String> = map.keys().collect();
            keys.sort();
            for key in keys {
                key.hash(hasher);
                hash_value(&map[key], hasher);
            }
        }
    }
}

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

    #[test]
    fn test_first_event_is_not_duplicate() {
        let store = DedupStore::new();
        let input = json!({"command": "ls -la"});

        let is_dup = store.check_and_insert("session-1", "bash", &input);

        assert!(!is_dup, "first occurrence must not be a duplicate");
    }

    #[test]
    fn test_same_event_within_ttl_is_duplicate() {
        let store = DedupStore::new();
        let input = json!({"command": "ls -la"});

        // First call — not a duplicate
        let first = store.check_and_insert("session-1", "bash", &input);
        // Second call immediately — within 100ms TTL
        let second = store.check_and_insert("session-1", "bash", &input);

        assert!(!first, "first occurrence must not be a duplicate");
        assert!(second, "immediate repeat must be detected as duplicate");
    }

    #[test]
    fn test_same_event_after_ttl_is_not_duplicate() {
        let store = DedupStore::new();
        let input = json!({"command": "ls -la"});

        store.check_and_insert("session-1", "bash", &input);

        // Wait beyond the 100ms TTL
        std::thread::sleep(Duration::from_millis(150));

        let after_ttl = store.check_and_insert("session-1", "bash", &input);
        assert!(!after_ttl, "event after TTL expiry must not be a duplicate");
    }

    #[test]
    fn test_different_events_are_not_duplicates() {
        let store = DedupStore::new();

        let first = store.check_and_insert("session-1", "bash", &json!({"cmd": "ls"}));
        let second =
            store.check_and_insert("session-1", "read_file", &json!({"path": "/etc/hosts"}));

        assert!(!first, "first event must not be a duplicate");
        assert!(!second, "different event must not be a duplicate");
    }

    #[test]
    fn test_different_sessions_same_tool_not_duplicate() {
        let store = DedupStore::new();
        let input = json!({"command": "ls"});

        store.check_and_insert("session-A", "bash", &input);
        let second = store.check_and_insert("session-B", "bash", &input);

        assert!(
            !second,
            "same tool call in different session must not be a duplicate"
        );
    }

    #[test]
    fn test_evict_expired_removes_old_entries() {
        let store = DedupStore::new();
        let input = json!({"key": "value"});

        store.check_and_insert("session-1", "bash", &input);

        // Confirm it's in the map (next call is a duplicate)
        assert!(
            store.check_and_insert("session-1", "bash", &input),
            "entry must exist before eviction"
        );

        // Wait for TTL to expire, then evict
        std::thread::sleep(Duration::from_millis(150));
        store.evict_expired();

        // After eviction, the same event is no longer a duplicate
        let after_evict = store.check_and_insert("session-1", "bash", &input);
        assert!(
            !after_evict,
            "evicted entry must not be a duplicate on next check"
        );
    }

    /// A call id stays claimed long past the content window, and names one
    /// call: another id, session or event type is not a duplicate of it.
    #[test]
    fn a_call_id_is_claimed_past_the_content_window() {
        let store = DedupStore::new();
        assert!(!store.check_and_insert_call("s", "pre_tool_use", "call_1"));
        std::thread::sleep(Duration::from_millis(150));
        store.evict_expired();
        assert!(store.check_and_insert_call("s", "pre_tool_use", "call_1"));
        assert!(!store.check_and_insert_call("s", "pre_tool_use", "call_2"));
        assert!(!store.check_and_insert_call("s", "post_tool_use", "call_1"));
        assert!(!store.check_and_insert_call("other", "pre_tool_use", "call_1"));
    }

    #[test]
    fn test_dedup_key_order_independent() {
        // Two JSON objects with the same keys in different insertion order
        // must produce the same dedup hash (canonical JSON sorts keys).
        let store = DedupStore::new();

        // Manually construct objects with different key order using serde_json::Map
        let mut map_a = serde_json::Map::new();
        map_a.insert("command".to_string(), json!("ls -la"));
        map_a.insert("path".to_string(), json!("/tmp"));
        let input_a = serde_json::Value::Object(map_a);

        let mut map_b = serde_json::Map::new();
        map_b.insert("path".to_string(), json!("/tmp"));
        map_b.insert("command".to_string(), json!("ls -la"));
        let input_b = serde_json::Value::Object(map_b);

        // First call with key order A
        let first = store.check_and_insert("session-1", "bash", &input_a);
        // Second call with key order B — must be detected as duplicate
        let second = store.check_and_insert("session-1", "bash", &input_b);

        assert!(!first, "first occurrence must not be a duplicate");
        assert!(
            second,
            "same logical event with different key order must be detected as duplicate"
        );
    }

    #[test]
    fn test_hash_value_sorts_nested_keys() {
        // Two objects with identical content but different key order must hash the same
        let input_a = json!({"z": {"b": 2, "a": 1}, "a": [{"y": 1, "x": 2}]});
        let input_b = json!({"a": [{"x": 2, "y": 1}], "z": {"a": 1, "b": 2}});

        let mut hasher_a = DefaultHasher::new();
        let mut hasher_b = DefaultHasher::new();
        super::hash_value(&input_a, &mut hasher_a);
        super::hash_value(&input_b, &mut hasher_b);

        assert_eq!(
            hasher_a.finish(),
            hasher_b.finish(),
            "nested keys in different order must produce the same hash"
        );
    }

    /// Claims count one-for-one, and ones nothing consumes expire on the next due prune instead
    /// of waiting for the hard cap.
    #[test]
    fn plugin_claims_count_and_expire() {
        let claims = PluginClaims::default();
        let call = json!({"tool": "read_file", "input": {"path": "/a"}});
        claims.insert("s", &call);
        claims.insert("s", &call);
        assert!(claims.consume("s", &call));
        assert!(claims.consume("s", &call));
        assert!(!claims.consume("s", &call), "two claims, two folds");
        assert_eq!(claims.len(), 0);

        claims.insert("cli", &call);
        claims.insert("cli", &json!({"prompt": "hi"}));
        assert_eq!(claims.len(), 2);
        claims.prune(Instant::now() + PLUGIN_CLAIM_TTL + Duration::from_secs(1));
        assert_eq!(
            claims.len(),
            0,
            "expired claims are dropped by time, not by count"
        );
    }
}