car_state/

lib.rs

//! State management for Common Agent Runtime.
//!
//! Provides structured, typed state with transition logging.
//! Every mutation produces a StateTransition record for audit and replay.
//!
//! ## Persistence (Parslee-ai/car#181)
//!
//! `StateStore::durable(path)` opens a JSONL-backed store. Each
//! mutation appends a transition line; on construction the file is
//! replayed to rebuild current state. This is the agent-persistence
//! pattern documented in `docs/persistence.md`. JSONL was chosen over
//! sqlite/sled to stay aligned with the existing JSONL persistence
//! used by `car-eventlog` and `car-memgine` — one file shape, one
//! reap+compact story, no native build deps.
//!
//! Per-key TTL is supported via `set_with_ttl` — the in-memory state
//! drops the key when `reap_expired(now)` runs after the deadline.
//! The on-disk file is compacted at the same time so the journal
//! doesn't grow unbounded.

use chrono::{DateTime, Duration, Utc};
use parking_lot::Mutex;
use serde::{Deserialize, Serialize};
use serde_json::Value;
use std::collections::HashMap;
use std::fs::{File, OpenOptions};
use std::io::{BufRead, BufReader, BufWriter, Write};
use std::path::{Path, PathBuf};

/// An explicit record of a state change.
///
/// `ttl_secs` is optional — when present, the key expires `ttl_secs`
/// seconds after `timestamp`. Reads return the value while it's
/// live; `reap_expired` drops it after the deadline. The default
/// (None) means "keep until explicitly deleted."
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct StateTransition {
    pub key: String,
    pub old_value: Option<Value>,
    pub new_value: Option<Value>,
    pub action_id: String,
    pub timestamp: DateTime<Utc>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub ttl_secs: Option<u64>,
    /// The key's monotonic version *after* this transition. Persisted so
    /// the version counter survives journal compaction and restart — a
    /// compacted journal collapses a key's history to one line, so without
    /// this field replay would recount from 1 and break the staleness
    /// guarantee (neo review M2). Optional for backward-compatible reads of
    /// pre-versioning journals.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub version: Option<u64>,
}

/// Thread-safe state store with transition logging.
///
/// All reads and writes go through this store. Every write produces a
/// StateTransition record for audit and replay. Optionally backed by
/// a JSONL journal file for durability across process restarts (see
/// [`StateStore::durable`]).
pub struct StateStore {
    state: Mutex<HashMap<String, Value>>,
    transitions: Mutex<Vec<StateTransition>>,
    /// Monotonic per-key version counter, bumped on every write/delete.
    /// The basis for transactional staleness detection (survey §5.2.4):
    /// an action that read `k` at version `v` can be flagged when `k` has
    /// since advanced past `v`, catching belief divergence that a value
    /// comparison alone would miss (e.g. set back to the same value).
    versions: Mutex<HashMap<String, u64>>,
    /// Optional JSONL-backed durability layer. When set, every
    /// `StateTransition` appended to the in-memory log is also
    /// appended to this file's open writer; `reap_expired` rewrites
    /// the file to compact away dropped keys.
    journal: Mutex<Option<Journal>>,
}

struct Journal {
    path: PathBuf,
    writer: BufWriter<File>,
}

impl StateStore {
    pub fn new() -> Self {
        Self {
            state: Mutex::new(HashMap::new()),
            transitions: Mutex::new(Vec::new()),
            versions: Mutex::new(HashMap::new()),
            journal: Mutex::new(None),
        }
    }

    /// Open a durable, JSONL-backed StateStore. If the file exists,
    /// its transitions are replayed (last-write-wins per key, with
    /// TTLs honored) to rebuild current state. Subsequent writes
    /// append to the same file.
    ///
    /// Returns an error only on filesystem-level failures (parent
    /// directory missing, permission denied, etc.). Malformed lines
    /// inside the journal are skipped with a warning rather than
    /// failing the open — agent persistence shouldn't refuse to
    /// start over a single bad line.
    pub fn durable(path: impl Into<PathBuf>) -> std::io::Result<Self> {
        let path = path.into();
        if let Some(parent) = path.parent() {
            if !parent.as_os_str().is_empty() {
                std::fs::create_dir_all(parent)?;
            }
        }
        let store = Self::new();
        store.replay_journal(&path)?;
        let file = OpenOptions::new().create(true).append(true).open(&path)?;
        *store.journal.lock() = Some(Journal {
            path,
            writer: BufWriter::new(file),
        });
        Ok(store)
    }

    fn replay_journal(&self, path: &Path) -> std::io::Result<()> {
        if !path.exists() {
            return Ok(());
        }
        let file = File::open(path)?;
        let reader = BufReader::new(file);
        let now = Utc::now();
        let mut state = self.state.lock();
        let mut transitions = self.transitions.lock();
        let mut versions = self.versions.lock();
        for line in reader.lines() {
            let line = match line {
                Ok(l) if l.trim().is_empty() => continue,
                Ok(l) => l,
                Err(_) => continue,
            };
            let Ok(t) = serde_json::from_str::<StateTransition>(&line) else {
                // Malformed line. Don't refuse to boot over it.
                tracing::warn!(
                    journal = %path.display(),
                    "skipping malformed StateStore journal line"
                );
                continue;
            };
            // Replay last-write-wins. TTLs that already expired are
            // dropped at replay time so we don't surface stale data
            // on first read.
            if let (Some(ttl), Some(value)) = (t.ttl_secs, &t.new_value) {
                if now.signed_duration_since(t.timestamp) > Duration::seconds(ttl as i64) {
                    state.remove(&t.key);
                } else {
                    state.insert(t.key.clone(), value.clone());
                }
            } else if let Some(value) = &t.new_value {
                state.insert(t.key.clone(), value.clone());
            } else {
                state.remove(&t.key);
            }
            // Restore the persisted version when present (survives
            // compaction); otherwise count transitions for legacy
            // pre-versioning journals. Take the max so the counter stays
            // monotonic across a mix of compacted and appended lines.
            let entry = versions.entry(t.key.clone()).or_insert(0);
            let restored = t.version.unwrap_or(*entry + 1);
            *entry = (*entry).max(restored);
            transitions.push(t);
        }
        Ok(())
    }

    fn append_journal(&self, transition: &StateTransition) {
        let mut journal = self.journal.lock();
        let Some(journal) = journal.as_mut() else {
            return;
        };
        // Best-effort: a failed disk write tracing::warn!s but the
        // in-memory write already succeeded. Callers who need
        // guaranteed durability should call `sync` after batches.
        let Ok(json) = serde_json::to_string(transition) else {
            return;
        };
        if let Err(e) = writeln!(journal.writer, "{json}") {
            tracing::warn!(
                journal = %journal.path.display(),
                error = %e,
                "StateStore journal append failed"
            );
            return;
        }
        let _ = journal.writer.flush();
    }

    /// Fsync the journal writer. Call after a batch of writes when
    /// you need durability guarantees beyond best-effort flush.
    pub fn sync(&self) -> std::io::Result<()> {
        let mut journal = self.journal.lock();
        let Some(journal) = journal.as_mut() else {
            return Ok(());
        };
        journal.writer.flush()?;
        journal.writer.get_ref().sync_all()
    }

    /// Drop expired keys (per `ttl_secs` on their last write) and
    /// rewrite the journal as a compacted snapshot of the surviving
    /// state. Returns the keys that were reaped.
    ///
    /// **TTL semantics**: a `ttl_secs` of 0 means "expired
    /// immediately" — the key is reapable on the next call. There
    /// is no "0 = forever" sentinel; use `set` (no TTL) for keys
    /// that should never auto-expire.
    ///
    /// Latest-write-wins: a key rewritten WITHOUT a TTL after a
    /// TTL'd write is NOT reaped — the more recent write
    /// effectively cancels the TTL.
    ///
    /// Single-pass over the transitions log via a key→latest
    /// index, so cost is O(n) in journal length (not O(n²)).
    pub fn reap_expired(&self, now: DateTime<Utc>) -> std::io::Result<Vec<String>> {
        let mut state = self.state.lock();
        let mut transitions = self.transitions.lock();
        // Build a single-pass index of the latest transition per
        // key. Walking the whole log once is unavoidable; doing it
        // ONCE keeps reap O(n) in journal length.
        let mut latest_by_key: HashMap<&str, &StateTransition> = HashMap::new();
        for t in transitions.iter() {
            latest_by_key.insert(t.key.as_str(), t);
        }
        let expired: Vec<String> = latest_by_key
            .iter()
            .filter_map(|(_k, t)| {
                let ttl = t.ttl_secs?;
                if t.new_value.is_none() {
                    return None;
                }
                let age = now.signed_duration_since(t.timestamp);
                (age > Duration::seconds(ttl as i64)).then(|| t.key.clone())
            })
            .collect();
        let mut reaped = Vec::new();
        for key in expired {
            if state.remove(&key).is_some() {
                // Bump the version so an expiry is observable as a change
                // (neo review N1: keeps in-memory versions in step with
                // what replay would reconstruct).
                let version = self.bump_version(&key);
                reaped.push(key.clone());
                transitions.push(StateTransition {
                    key,
                    old_value: None,
                    new_value: None,
                    action_id: "reap".to_string(),
                    timestamp: now,
                    ttl_secs: None,
                    version: Some(version),
                });
            }
        }
        drop(state);
        drop(transitions);
        if !reaped.is_empty() {
            self.compact_journal()?;
        }
        Ok(reaped)
    }

    /// Rewrite the journal as a flat snapshot of the current state —
    /// one transition per surviving key, no replay history. Reduces
    /// journal size without changing observable behavior.
    ///
    /// **Concurrency requirement**: callers MUST hold the
    /// observation that no other thread is mid-`set`/`delete` on
    /// this store; the snapshot is taken under the state lock, but
    /// in-flight journal appends to the *old* file handle that
    /// land between snapshot and rename are lost. Today's only
    /// caller is `reap_expired`, which holds both locks across the
    /// call; external callers should serialize themselves.
    pub(crate) fn compact_journal(&self) -> std::io::Result<()> {
        let mut journal = self.journal.lock();
        let Some(j) = journal.as_mut() else {
            return Ok(());
        };
        let state = self.state.lock().clone();
        let versions = self.versions.lock().clone();
        let tmp_path = j.path.with_extension("jsonl.tmp");
        {
            let tmp_file = File::create(&tmp_path)?;
            let mut writer = BufWriter::new(tmp_file);
            for (key, value) in &state {
                let t = StateTransition {
                    key: key.clone(),
                    old_value: None,
                    new_value: Some(value.clone()),
                    action_id: "compact".to_string(),
                    timestamp: Utc::now(),
                    ttl_secs: None,
                    // Persist the true version so replay restores it rather
                    // than recounting the collapsed history from 1.
                    version: versions.get(key).copied(),
                };
                let line = serde_json::to_string(&t)?;
                writeln!(writer, "{line}")?;
            }
            writer.flush()?;
            writer.get_ref().sync_all()?;
        }
        std::fs::rename(&tmp_path, &j.path)?;
        let file = OpenOptions::new().create(true).append(true).open(&j.path)?;
        j.writer = BufWriter::new(file);
        Ok(())
    }

    pub fn get(&self, key: &str) -> Option<Value> {
        self.state.lock().get(key).cloned()
    }

    pub fn get_or(&self, key: &str, default: Value) -> Value {
        self.state.lock().get(key).cloned().unwrap_or(default)
    }

    pub fn exists(&self, key: &str) -> bool {
        self.state.lock().contains_key(key)
    }

    pub fn set(&self, key: &str, value: Value, action_id: &str) -> StateTransition {
        self.set_inner(key, value, action_id, None)
    }

    /// Set a key with a TTL (seconds from now). `reap_expired`
    /// drops the key once the deadline passes; re-setting the key
    /// without a TTL (`set`) cancels the TTL.
    ///
    /// `ttl_secs == 0` means "expire immediately" (reapable on the
    /// next `reap_expired` call). It is NOT a "no expiry" sentinel
    /// — use the plain `set(...)` method for keys that should
    /// never auto-expire. This differs from the Unix/Redis
    /// convention; the distinction matters because a TTL passed
    /// from untrusted input could otherwise silently mean
    /// "forever" when the caller intended "never store."
    pub fn set_with_ttl(
        &self,
        key: &str,
        value: Value,
        action_id: &str,
        ttl_secs: u64,
    ) -> StateTransition {
        self.set_inner(key, value, action_id, Some(ttl_secs))
    }

    fn set_inner(
        &self,
        key: &str,
        value: Value,
        action_id: &str,
        ttl_secs: Option<u64>,
    ) -> StateTransition {
        let mut state = self.state.lock();
        let old = state.get(key).cloned();
        state.insert(key.to_string(), value.clone());
        let version = self.bump_version(key);

        let t = StateTransition {
            key: key.to_string(),
            old_value: old,
            new_value: Some(value),
            action_id: action_id.to_string(),
            timestamp: Utc::now(),
            ttl_secs,
            version: Some(version),
        };

        self.transitions.lock().push(t.clone());
        self.append_journal(&t);
        t
    }

    /// Increment the monotonic version counter for `key`, returning the new
    /// version.
    fn bump_version(&self, key: &str) -> u64 {
        let mut versions = self.versions.lock();
        let v = versions.entry(key.to_string()).or_insert(0);
        *v += 1;
        *v
    }

    /// Current version of `key` (number of writes/deletes applied to it),
    /// or `None` if it was never written. Used by the transactional
    /// conflict checker to detect stale reads (survey §5.2.4).
    pub fn version(&self, key: &str) -> Option<u64> {
        self.versions.lock().get(key).copied()
    }

    /// Snapshot of all key versions — the version map an action's
    /// assumptions are checked against.
    pub fn versions(&self) -> HashMap<String, u64> {
        self.versions.lock().clone()
    }

    /// Atomic snapshot of both the current values and the current versions,
    /// taken under a single consistent lock acquisition (state then
    /// versions, matching the write path) so the two maps can't tear — a
    /// caller never sees a value from version N+1 paired with version N
    /// (neo review N2). This is the pair the transactional conflict checker
    /// (`car_verify::check_transaction`) should consume.
    pub fn versioned_snapshot(&self) -> (HashMap<String, Value>, HashMap<String, u64>) {
        let state = self.state.lock();
        let versions = self.versions.lock();
        (state.clone(), versions.clone())
    }

    pub fn delete(&self, key: &str, action_id: &str) -> Option<StateTransition> {
        let mut state = self.state.lock();
        let old = state.remove(key)?;
        let version = self.bump_version(key);

        let t = StateTransition {
            key: key.to_string(),
            old_value: Some(old),
            new_value: None,
            action_id: action_id.to_string(),
            timestamp: Utc::now(),
            ttl_secs: None,
            version: Some(version),
        };

        self.transitions.lock().push(t.clone());
        self.append_journal(&t);
        Some(t)
    }

    /// Deep clone of current state.
    pub fn snapshot(&self) -> HashMap<String, Value> {
        self.state.lock().clone()
    }

    /// Restore state from a snapshot, truncating transitions.
    pub fn restore(&self, snapshot: HashMap<String, Value>, transition_count: usize) {
        *self.state.lock() = snapshot;
        self.transitions.lock().truncate(transition_count);
    }

    pub fn transition_count(&self) -> usize {
        self.transitions.lock().len()
    }

    pub fn transitions(&self) -> Vec<StateTransition> {
        self.transitions.lock().clone()
    }

    pub fn transitions_since(&self, index: usize) -> Vec<StateTransition> {
        let transitions = self.transitions.lock();
        let start = index.min(transitions.len());
        transitions[start..].to_vec()
    }

    pub fn keys(&self) -> Vec<String> {
        self.state.lock().keys().cloned().collect()
    }

    /// Replace the entire state map without recording transitions.
    /// Used by checkpoint restore to avoid synthetic transition history.
    /// Also clears the transitions log so callers of `transitions_since()`
    /// don't see stale history from the discarded state.
    pub fn replace_all(&self, snapshot: HashMap<String, Value>) {
        *self.state.lock() = snapshot;
        self.transitions.lock().clear();
    }

    /// Build a tenant-scoped view over this store
    /// (Parslee-ai/car#187 phase 3 enforcement).
    ///
    /// All reads / writes go through `tenant:<tenant_id>:<key>` so
    /// distinct tenants can't see each other's keys. `tenant = None`
    /// returns a view that hits the unscoped (legacy) namespace —
    /// callers that don't yet have a `RuntimeScope` get pre-#187
    /// behaviour automatically.
    ///
    /// Cheap to construct; holds a `&self` borrow plus the tenant
    /// string. The view's methods take the parking-lot lock the same
    /// way the unscoped methods do.
    pub fn scoped<'a>(&'a self, tenant: Option<&'a str>) -> ScopedStateView<'a> {
        ScopedStateView {
            store: self,
            tenant,
        }
    }
}

/// Tenant-scoped view over a [`StateStore`]. All key arguments are
/// transparently prefixed with `tenant:<tenant_id>:` before hitting
/// the underlying store; on the way out, the prefix is stripped so
/// callers see their original keys.
///
/// Construct via [`StateStore::scoped`]. When `tenant` is `None`,
/// the prefix is empty and the view is functionally equivalent to
/// the unscoped methods on `StateStore` — useful for code paths
/// that always go through this view regardless of whether scope is
/// active.
///
/// # Isolation guarantee
///
/// Two views with distinct `tenant` strings cannot observe each
/// other's writes through `get` / `exists` / `keys`. The transitions
/// log still records the full (prefixed) key so audit / replay sees
/// the actual storage layout.
///
/// # What isolation does *not* cover (phase 3 follow-ups)
///
/// - `StateStore::snapshot` / `restore` are deliberately unscoped —
///   they're called at proposal boundaries for rollback and need to
///   see the whole map. Per-tenant partial rollback is a known
///   concurrency hole when multiple proposals run interleaved; the
///   pre-#187 baseline has the same issue, and fixing it cleanly
///   requires either serializing per-tenant or extending the
///   transactional model. Tracked as a follow-up.
/// - The journal file (when durability is on) records full
///   prefixed keys. Operators rotating tenants out can grep the
///   journal by prefix.
pub struct ScopedStateView<'a> {
    store: &'a StateStore,
    tenant: Option<&'a str>,
}

impl<'a> ScopedStateView<'a> {
    fn full_key(&self, key: &str) -> String {
        match self.tenant {
            Some(t) if !t.is_empty() => format!("tenant:{t}:{key}"),
            _ => key.to_string(),
        }
    }

    fn strip_prefix<'k>(&self, full: &'k str) -> Option<&'k str> {
        match self.tenant {
            Some(t) if !t.is_empty() => {
                let prefix = format!("tenant:{t}:");
                full.strip_prefix(&prefix)
            }
            _ => Some(full),
        }
    }

    pub fn get(&self, key: &str) -> Option<Value> {
        self.store.get(&self.full_key(key))
    }

    pub fn get_or(&self, key: &str, default: Value) -> Value {
        self.store.get_or(&self.full_key(key), default)
    }

    pub fn exists(&self, key: &str) -> bool {
        self.store.exists(&self.full_key(key))
    }

    pub fn set(&self, key: &str, value: Value, action_id: &str) -> StateTransition {
        self.store.set(&self.full_key(key), value, action_id)
    }

    pub fn set_with_ttl(
        &self,
        key: &str,
        value: Value,
        action_id: &str,
        ttl_secs: u64,
    ) -> StateTransition {
        self.store
            .set_with_ttl(&self.full_key(key), value, action_id, ttl_secs)
    }

    pub fn delete(&self, key: &str, action_id: &str) -> Option<StateTransition> {
        self.store.delete(&self.full_key(key), action_id)
    }

    /// Return keys belonging to this tenant only, with the
    /// `tenant:<id>:` prefix stripped so callers see their original
    /// key names. Unscoped views (no tenant) return only keys that
    /// don't start with `tenant:` — preventing accidental visibility
    /// of scoped state through a legacy code path.
    pub fn keys(&self) -> Vec<String> {
        self.store
            .keys()
            .into_iter()
            .filter_map(|k| {
                if self.tenant.map(|t| !t.is_empty()).unwrap_or(false) {
                    self.strip_prefix(&k).map(str::to_string)
                } else if k.starts_with("tenant:") {
                    None
                } else {
                    Some(k)
                }
            })
            .collect()
    }
}

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

impl car_ir::precondition::StateView for StateStore {
    fn get_value(&self, key: &str) -> Option<Value> {
        self.get(key)
    }
    fn key_exists(&self, key: &str) -> bool {
        self.exists(key)
    }
}

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

    #[test]
    fn set_and_get() {
        let store = StateStore::new();
        store.set("x", Value::from(42), "test");
        assert_eq!(store.get("x"), Some(Value::from(42)));
    }

    #[test]
    fn exists() {
        let store = StateStore::new();
        assert!(!store.exists("x"));
        store.set("x", Value::from(1), "test");
        assert!(store.exists("x"));
    }

    #[test]
    fn delete() {
        let store = StateStore::new();
        store.set("x", Value::from(1), "test");
        let t = store.delete("x", "test");
        assert!(t.is_some());
        assert!(!store.exists("x"));
    }

    #[test]
    fn delete_nonexistent() {
        let store = StateStore::new();
        assert!(store.delete("x", "test").is_none());
    }

    #[test]
    fn snapshot_and_restore() {
        let store = StateStore::new();
        store.set("x", Value::from(1), "a");
        let snap = store.snapshot();
        let tc = store.transition_count();

        store.set("y", Value::from(2), "b");
        assert!(store.exists("y"));

        store.restore(snap, tc);
        assert!(store.exists("x"));
        assert!(!store.exists("y"));
        assert_eq!(store.transition_count(), 1);
    }

    #[test]
    fn transitions_logged() {
        let store = StateStore::new();
        store.set("a", Value::from(1), "act1");
        store.set("b", Value::from(2), "act2");

        let transitions = store.transitions();
        assert_eq!(transitions.len(), 2);
        assert_eq!(transitions[0].key, "a");
        assert_eq!(transitions[1].key, "b");
    }

    #[test]
    fn transitions_since() {
        let store = StateStore::new();
        store.set("a", Value::from(1), "act1");
        let idx = store.transition_count();
        store.set("b", Value::from(2), "act2");

        let since = store.transitions_since(idx);
        assert_eq!(since.len(), 1);
        assert_eq!(since[0].key, "b");
    }

    #[test]
    fn transition_records_old_value() {
        let store = StateStore::new();
        store.set("x", Value::from(1), "first");
        store.set("x", Value::from(2), "second");

        let transitions = store.transitions();
        assert_eq!(transitions[1].old_value, Some(Value::from(1)));
        assert_eq!(transitions[1].new_value, Some(Value::from(2)));
    }

    #[test]
    fn keys() {
        let store = StateStore::new();
        store.set("a", Value::from(1), "t");
        store.set("b", Value::from(2), "t");
        let mut keys = store.keys();
        keys.sort();
        assert_eq!(keys, vec!["a", "b"]);
    }

    #[test]
    fn transitions_since_after_restore_does_not_panic() {
        let store = StateStore::new();
        store.set("a", serde_json::json!(1), "test");
        store.set("b", serde_json::json!(2), "test");
        let count_before = store.transition_count(); // 2

        // Restore to empty, truncating transitions to 0
        store.restore(HashMap::new(), 0);

        // Using the stale count_before (2) should not panic
        let result = store.transitions_since(count_before);
        assert!(result.is_empty());
    }

    #[test]
    fn transitions_since_normal_usage() {
        let store = StateStore::new();
        store.set("a", serde_json::json!(1), "test");
        let mark = store.transition_count();
        store.set("b", serde_json::json!(2), "test");
        let since = store.transitions_since(mark);
        assert_eq!(since.len(), 1);
        assert_eq!(since[0].key, "b");
    }

    #[test]
    fn replace_all_swaps_state_without_transitions() {
        let store = StateStore::new();
        store.set("old_key", serde_json::json!("old"), "setup");

        let mut new_state = HashMap::new();
        new_state.insert("new_key".to_string(), serde_json::json!("new"));
        store.replace_all(new_state);

        assert_eq!(store.get("new_key"), Some(serde_json::json!("new")));
        assert_eq!(store.get("old_key"), None);
        // After replace_all, transitions should be cleared (not preserved)
        assert_eq!(store.transition_count(), 0);
    }

    #[test]
    fn durable_store_survives_reopen() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("state.jsonl");
        {
            let store = StateStore::durable(&path).unwrap();
            store.set("agent", serde_json::json!("planner"), "boot");
            store.set("turns", serde_json::json!(42), "tick");
            store.sync().unwrap();
        }
        let store = StateStore::durable(&path).unwrap();
        assert_eq!(store.get("agent"), Some(serde_json::json!("planner")));
        assert_eq!(store.get("turns"), Some(serde_json::json!(42)));
    }

    #[test]
    fn durable_store_replays_deletes() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("state.jsonl");
        {
            let store = StateStore::durable(&path).unwrap();
            store.set("transient", serde_json::json!("x"), "boot");
            store.delete("transient", "rm");
            store.sync().unwrap();
        }
        let store = StateStore::durable(&path).unwrap();
        assert!(!store.exists("transient"));
    }

    #[test]
    fn ttl_reap_drops_expired_and_keeps_fresh() {
        let store = StateStore::new();
        store.set_with_ttl("short", serde_json::json!(1), "set", 0);
        store.set_with_ttl("long", serde_json::json!(2), "set", 3600);
        store.set("forever", serde_json::json!(3), "set");
        // Now + 10s — short (ttl=0) is expired, long (ttl=3600) is fresh, forever has no TTL.
        let reaped = store
            .reap_expired(Utc::now() + Duration::seconds(10))
            .unwrap();
        assert_eq!(reaped, vec!["short".to_string()]);
        assert!(!store.exists("short"));
        assert_eq!(store.get("long"), Some(serde_json::json!(2)));
        assert_eq!(store.get("forever"), Some(serde_json::json!(3)));
    }

    #[test]
    fn durable_ttl_compacts_journal() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("state.jsonl");
        {
            let store = StateStore::durable(&path).unwrap();
            for i in 0..50 {
                store.set_with_ttl(&format!("k{i}"), serde_json::json!(i), "set", 0);
            }
            store.set("survivor", serde_json::json!("kept"), "set");
            store.sync().unwrap();
            let pre = std::fs::metadata(&path).unwrap().len();
            // Force expiry by advancing the clock past the 0s TTL.
            let reaped = store
                .reap_expired(Utc::now() + Duration::seconds(1))
                .unwrap();
            assert_eq!(reaped.len(), 50);
            store.sync().unwrap();
            let post = std::fs::metadata(&path).unwrap().len();
            // Compaction should shrink the journal: 50 TTL'd writes + 1
            // survivor pre-compact is 51 lines; post-compact is 1 line.
            assert!(
                post < pre,
                "post={post} pre={pre} — compaction did not shrink"
            );
        }
        // Reopen — only the survivor remains.
        let store = StateStore::durable(&path).unwrap();
        assert!(!store.exists("k0"));
        assert!(!store.exists("k49"));
        assert_eq!(store.get("survivor"), Some(serde_json::json!("kept")));
        // Version survives compaction (neo M2): survivor was written once,
        // so its version is 1 after a compaction-then-reopen, not reset in
        // a way that breaks staleness detection.
        assert_eq!(store.version("survivor"), Some(1));
    }

    #[test]
    fn version_is_monotonic_and_survives_compaction() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("v.jsonl");
        {
            let store = StateStore::durable(&path).unwrap();
            for i in 0..3 {
                store.set("cfg", serde_json::json!(i), "set");
            }
            assert_eq!(store.version("cfg"), Some(3));
            // A TTL key that expires forces a compaction of the journal.
            store.set_with_ttl("tmp", serde_json::json!(1), "set", 0);
            store.sync().unwrap();
            store.reap_expired(Utc::now() + Duration::seconds(1)).unwrap();
            store.sync().unwrap();
        }
        // After compaction + restart, cfg's version must still be 3 — not
        // recounted to 1 from the collapsed single line.
        let store = StateStore::durable(&path).unwrap();
        assert_eq!(store.version("cfg"), Some(3));
    }

    #[test]
    fn reap_bumps_version() {
        let store = StateStore::new();
        store.set("k", serde_json::json!("v"), "set");
        assert_eq!(store.version("k"), Some(1));
        store.set_with_ttl("k", serde_json::json!("v2"), "set", 0);
        assert_eq!(store.version("k"), Some(2));
        store.reap_expired(Utc::now() + Duration::seconds(1)).unwrap();
        // Expiry is an observable change → version advances (neo N1).
        assert_eq!(store.version("k"), Some(3));
    }

    #[test]
    fn ttl_then_rewrite_without_ttl_does_not_reap() {
        let store = StateStore::new();
        store.set_with_ttl("k", serde_json::json!("a"), "first", 0);
        store.set("k", serde_json::json!("b"), "second"); // no TTL
        let reaped = store
            .reap_expired(Utc::now() + Duration::seconds(10))
            .unwrap();
        assert!(reaped.is_empty());
        assert_eq!(store.get("k"), Some(serde_json::json!("b")));
    }

    #[test]
    fn malformed_journal_line_is_skipped_not_fatal() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("state.jsonl");
        // Plant a good line + a bad line + another good line.
        {
            std::fs::write(
                &path,
                "{\"key\":\"a\",\"old_value\":null,\"new_value\":1,\"action_id\":\"x\",\"timestamp\":\"2026-05-11T00:00:00Z\"}\n\
                 not-json\n\
                 {\"key\":\"b\",\"old_value\":null,\"new_value\":2,\"action_id\":\"x\",\"timestamp\":\"2026-05-11T00:00:00Z\"}\n",
            )
            .unwrap();
        }
        let store = StateStore::durable(&path).unwrap();
        assert_eq!(store.get("a"), Some(serde_json::json!(1)));
        assert_eq!(store.get("b"), Some(serde_json::json!(2)));
    }

    // ScopedStateView tests — Parslee-ai/car#187 phase 3 enforcement.

    #[test]
    fn scoped_view_writes_isolate_between_tenants() {
        let store = StateStore::new();
        store.scoped(Some("acme")).set("config", json!("A"), "act");
        store
            .scoped(Some("globex"))
            .set("config", json!("G"), "act");

        // Each tenant sees their own value.
        assert_eq!(store.scoped(Some("acme")).get("config"), Some(json!("A")));
        assert_eq!(store.scoped(Some("globex")).get("config"), Some(json!("G")));
    }

    #[test]
    fn scoped_view_isolates_existence_check() {
        let store = StateStore::new();
        store.scoped(Some("acme")).set("k", json!(1), "act");
        assert!(store.scoped(Some("acme")).exists("k"));
        assert!(!store.scoped(Some("globex")).exists("k"));
    }

    #[test]
    fn scoped_view_keys_filters_to_tenant() {
        let store = StateStore::new();
        store.scoped(Some("acme")).set("a", json!(1), "act");
        store.scoped(Some("acme")).set("b", json!(2), "act");
        store.scoped(Some("globex")).set("g", json!(9), "act");
        store.set("unscoped", json!(0), "act");

        let mut acme_keys = store.scoped(Some("acme")).keys();
        acme_keys.sort();
        assert_eq!(acme_keys, vec!["a", "b"]);

        let globex_keys = store.scoped(Some("globex")).keys();
        assert_eq!(globex_keys, vec!["g"]);
    }

    #[test]
    fn unscoped_view_skips_tenant_prefixed_keys() {
        // Calling scoped(None) — the legacy-compat path — must NOT
        // accidentally expose other tenants' keys via `keys()`. This
        // is the inverse of the isolation contract: the unscoped
        // namespace shouldn't see scoped data even though it's all
        // in the same backing HashMap.
        let store = StateStore::new();
        store.set("legacy", json!("ok"), "act");
        store.scoped(Some("acme")).set("hidden", json!(42), "act");

        let unscoped = store.scoped(None).keys();
        assert_eq!(unscoped, vec!["legacy"]);
        assert!(store.scoped(None).get("hidden").is_none());
    }

    #[test]
    fn scoped_view_delete_doesnt_touch_other_tenants() {
        let store = StateStore::new();
        store.scoped(Some("acme")).set("shared", json!(1), "act");
        store.scoped(Some("globex")).set("shared", json!(2), "act");

        store.scoped(Some("acme")).delete("shared", "act");
        assert!(!store.scoped(Some("acme")).exists("shared"));
        assert!(store.scoped(Some("globex")).exists("shared"));
    }

    #[test]
    fn empty_tenant_string_treated_as_unscoped() {
        // Some(""): defensive — RuntimeScope normalizes empty strings
        // to None at the dispatcher, but the view shouldn't trip if
        // a caller passes an empty tenant by mistake.
        let store = StateStore::new();
        store.scoped(Some("")).set("k", json!(1), "act");
        assert_eq!(store.get("k"), Some(json!(1)));
        assert_eq!(store.scoped(None).get("k"), Some(json!(1)));
    }
}