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.

pub mod crdt;

use chrono::{DateTime, Duration, Utc};
use parking_lot::{Mutex, MutexGuard};
use serde::{Deserialize, Serialize};
use serde_json::Value;
use std::collections::HashMap;
use std::ffi::OsString;
use std::fs::{File, OpenOptions};
use std::io::{BufRead, BufReader, BufWriter, Write};
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::Arc;
use std::time::{SystemTime, UNIX_EPOCH};

/// Durability reached by a persistence-aware state restore.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum RestoreDurability {
    /// The replacement journal and its parent directory were fsynced.
    Durable,
    /// The replacement journal is visible and memory adopted it, but the
    /// parent-directory metadata flush failed, so crash persistence is unknown.
    DurabilityUnknown { error: String },
}

/// Deterministic durability fault seam for restore integration tests.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum RestoreFailurePoint {
    BeforePublication,
    ParentDirectorySync,
}

#[derive(Debug, Clone, Default)]
pub struct RestoreFailureInjector {
    failures: Arc<std::sync::Mutex<Vec<RestoreFailurePoint>>>,
}

impl RestoreFailureInjector {
    pub fn fail_next(&self, point: RestoreFailurePoint) {
        self.failures
            .lock()
            .expect("restore failure injector lock poisoned")
            .push(point);
    }

    fn check(&self, point: RestoreFailurePoint) -> std::io::Result<()> {
        let mut failures = self
            .failures
            .lock()
            .map_err(|_| std::io::Error::other("restore failure injector lock poisoned"))?;
        if failures.first() == Some(&point) {
            failures.remove(0);
            let message = match point {
                RestoreFailurePoint::BeforePublication => {
                    "injected restore failure before publication"
                }
                RestoreFailurePoint::ParentDirectorySync => {
                    "injected restore parent directory sync failure"
                }
            };
            return Err(std::io::Error::other(message));
        }
        Ok(())
    }
}

/// 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>,
}

/// What `replay_journal` found at the end of the file, and what `durable`
/// must do about it before opening the append writer. An interrupted append
/// can leave the final line unterminated; resuming appends on that tail
/// would merge two records into one malformed line that the NEXT reopen
/// drops whole (Parslee-ai/car#1140).
enum TailRepair {
    /// File ends cleanly (or doesn't exist).
    None,
    /// The final record is complete JSON but its newline is missing — it
    /// was replayed; write the terminator so the next append starts fresh.
    Terminate,
    /// The final line is a torn prefix — it was skipped; truncate the file
    /// to this byte offset so the next append cannot merge with it.
    TruncateTo(u64),
}

/// One journal line holding every transition of a single atomic batch
/// mutation ([`StateStore::set_batch`]). A multi-key batch is journaled as
/// one line so an interrupted append can never persist a prefix of the
/// batch: replay applies the complete record, or — when the line is torn —
/// skips it whole via the existing malformed-line path. Old-or-complete,
/// never partial (Parslee-ai/car#1140). Single-entry batches keep the
/// legacy bare-`StateTransition` line shape, so a journal only contains
/// this record where a multi-key batch actually occurred.
#[derive(Debug, Deserialize)]
struct BatchTransitionRecord {
    batch: Vec<StateTransition>,
}

/// Borrowed serialization twin of [`BatchTransitionRecord`] — writes the
/// same `{"batch": [...]}` shape without cloning the transitions.
#[derive(Serialize)]
struct BatchTransitionRecordRef<'a> {
    batch: &'a [StateTransition],
}

/// 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 {
    /// Serializes proposal transactions across every Runtime sharing this
    /// store. State transitions carry action ids for compatibility, so the
    /// shared store—not an individual Runtime—is the safe boundary that keeps
    /// reused ids from cross-attributing concurrent proposals.
    proposal_execution: tokio::sync::Mutex<()>,
    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>>,
    restore_failures: Option<RestoreFailureInjector>,
    #[cfg(test)]
    mutation_before_state_lock: Option<Arc<MutationRaceBarrier>>,
    /// Test seam: pause a [`Self::set_batch`] after its first entry is
    /// applied, with the state lock still held — lets a test prove a
    /// concurrent reader blocks rather than observing a batch prefix.
    #[cfg(test)]
    batch_mid_apply: Option<Arc<MutationRaceBarrier>>,
}

#[cfg(test)]
struct MutationRaceBarrier {
    reached: std::sync::Barrier,
    release: std::sync::Barrier,
}

#[cfg(test)]
impl MutationRaceBarrier {
    fn new() -> Self {
        Self {
            reached: std::sync::Barrier::new(2),
            release: std::sync::Barrier::new(2),
        }
    }

    fn pause_mutation(&self) {
        self.reached.wait();
        self.release.wait();
    }
}

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

impl Journal {
    fn reopen_writer(&mut self) -> std::io::Result<()> {
        let file = OpenOptions::new().append(true).open(&self.path)?;
        self.writer = Some(BufWriter::new(file));
        Ok(())
    }

    fn writer_mut(&mut self) -> std::io::Result<&mut BufWriter<File>> {
        if self.writer.is_none() {
            self.reopen_writer()?;
        }
        self.writer
            .as_mut()
            .ok_or_else(|| std::io::Error::other("state journal writer is unavailable"))
    }
}

const REPLACEMENT_TEMP_ATTEMPTS: usize = 32;
static NEXT_REPLACEMENT_TEMP_ID: AtomicU64 = AtomicU64::new(0);

struct TempFileCleanup {
    path: PathBuf,
    armed: bool,
}

impl TempFileCleanup {
    fn new(path: PathBuf) -> Self {
        Self { path, armed: true }
    }

    fn disarm(&mut self) {
        self.armed = false;
    }
}

impl Drop for TempFileCleanup {
    fn drop(&mut self) {
        if self.armed {
            let _ = std::fs::remove_file(&self.path);
        }
    }
}

fn create_replacement_temp(destination: &Path) -> std::io::Result<(PathBuf, File)> {
    let parent = destination.parent().unwrap_or_else(|| Path::new("."));
    let file_name = destination.file_name().ok_or_else(|| {
        std::io::Error::new(
            std::io::ErrorKind::InvalidInput,
            "state journal path has no file name",
        )
    })?;
    let timestamp = SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .unwrap_or_default()
        .as_nanos();

    for _ in 0..REPLACEMENT_TEMP_ATTEMPTS {
        let id = NEXT_REPLACEMENT_TEMP_ID.fetch_add(1, Ordering::Relaxed);
        let mut temp_name = OsString::from(".");
        temp_name.push(file_name);
        temp_name.push(format!(
            ".restore.{}.{}.{}.tmp",
            std::process::id(),
            timestamp,
            id
        ));
        let temp_path = parent.join(temp_name);
        let mut options = OpenOptions::new();
        options.write(true).create_new(true);
        #[cfg(unix)]
        {
            use std::os::unix::fs::OpenOptionsExt;
            options.mode(0o600);
        }
        match options.open(&temp_path) {
            Ok(file) => return Ok((temp_path, file)),
            Err(error) if error.kind() == std::io::ErrorKind::AlreadyExists => continue,
            Err(error) => return Err(error),
        }
    }

    Err(std::io::Error::new(
        std::io::ErrorKind::AlreadyExists,
        format!(
            "could not allocate a unique state journal replacement after {REPLACEMENT_TEMP_ATTEMPTS} attempts"
        ),
    ))
}

#[cfg(unix)]
fn replace_file_atomically(temp: &Path, destination: &Path) -> std::io::Result<()> {
    std::fs::rename(temp, destination)
}

#[cfg(target_os = "windows")]
fn replace_file_atomically(temp: &Path, destination: &Path) -> std::io::Result<()> {
    use std::os::windows::ffi::OsStrExt;
    use windows::core::PCWSTR;
    use windows::Win32::Storage::FileSystem::{
        MoveFileExW, MOVEFILE_REPLACE_EXISTING, MOVEFILE_WRITE_THROUGH,
    };

    let temp: Vec<u16> = temp.as_os_str().encode_wide().chain(Some(0)).collect();
    let destination: Vec<u16> = destination
        .as_os_str()
        .encode_wide()
        .chain(Some(0))
        .collect();
    unsafe {
        MoveFileExW(
            PCWSTR(temp.as_ptr()),
            PCWSTR(destination.as_ptr()),
            MOVEFILE_REPLACE_EXISTING | MOVEFILE_WRITE_THROUGH,
        )
    }
    .map_err(|error| std::io::Error::other(error.to_string()))
}

#[cfg(not(any(unix, target_os = "windows")))]
fn replace_file_atomically(temp: &Path, destination: &Path) -> std::io::Result<()> {
    std::fs::rename(temp, destination)
}

#[cfg(unix)]
fn sync_parent_directory(path: &Path) -> std::io::Result<()> {
    File::open(path.parent().unwrap_or_else(|| Path::new(".")))?.sync_all()
}

#[cfg(target_os = "windows")]
fn sync_parent_directory(_path: &Path) -> std::io::Result<()> {
    // Windows does not expose a supported directory-fsync equivalent.
    // Publication uses MoveFileExW(MOVEFILE_WRITE_THROUGH), which does not
    // return until the move has reached durable storage, so there is no
    // additional parent-directory handle to flush here.
    Ok(())
}

#[cfg(not(any(unix, target_os = "windows")))]
fn sync_parent_directory(_path: &Path) -> std::io::Result<()> {
    Ok(())
}

impl StateStore {
    pub fn new() -> Self {
        Self {
            proposal_execution: tokio::sync::Mutex::new(()),
            state: Mutex::new(HashMap::new()),
            transitions: Mutex::new(Vec::new()),
            versions: Mutex::new(HashMap::new()),
            journal: Mutex::new(None),
            restore_failures: None,
            #[cfg(test)]
            mutation_before_state_lock: None,
            #[cfg(test)]
            batch_mid_apply: None,
        }
    }

    /// Shared proposal-transaction guard. Hold this across proposal execution,
    /// including rollback and replanning; actions inside the guarded proposal
    /// may still execute concurrently according to their DAG.
    pub async fn lock_proposal_execution(&self) -> tokio::sync::MutexGuard<'_, ()> {
        self.proposal_execution.lock().await
    }

    /// 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> {
        Self::durable_with_restore_failure_injector(path, None)
    }

    #[doc(hidden)]
    pub fn durable_with_restore_failure_injector(
        path: impl Into<PathBuf>,
        restore_failures: impl Into<Option<RestoreFailureInjector>>,
    ) -> 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 mut store = Self::new();
        store.restore_failures = restore_failures.into();
        match store.replay_journal(&path)? {
            TailRepair::None => {}
            // The final record is complete but its newline never made it to
            // disk. It was replayed above; terminate it so the next append
            // starts a fresh line instead of gluing onto it.
            TailRepair::Terminate => {
                let mut file = OpenOptions::new().append(true).open(&path)?;
                file.write_all(b"\n")?;
                file.sync_all()?;
            }
            // The final line is a torn prefix of an interrupted append. It
            // was skipped above (old-or-complete); truncate it away so the
            // next append cannot merge with it into one malformed line that
            // a later reopen would then drop along with the new record.
            TailRepair::TruncateTo(offset) => {
                let file = OpenOptions::new().write(true).open(&path)?;
                file.set_len(offset)?;
                file.sync_all()?;
            }
        }
        let file = OpenOptions::new().create(true).append(true).open(&path)?;
        *store.journal.lock() = Some(Journal {
            path,
            writer: Some(BufWriter::new(file)),
            pending_parent_sync: None,
        });
        Ok(store)
    }

    fn replay_journal(&self, path: &Path) -> std::io::Result<TailRepair> {
        if !path.exists() {
            return Ok(TailRepair::None);
        }
        let file = File::open(path)?;
        let mut 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();
        let mut buf = Vec::new();
        let mut offset: u64 = 0;
        let mut repair = TailRepair::None;
        loop {
            buf.clear();
            let read = match reader.read_until(b'\n', &mut buf) {
                Ok(0) => break,
                Ok(n) => n,
                Err(_) => break,
            };
            let line_start = offset;
            offset += read as u64;
            // A process can crash after writing only part of its final
            // JSONL row — including everything but the trailing newline.
            // The append writer resumes at EOF, so an unrepaired tail
            // would glue the NEXT record onto this one into a single
            // malformed line that a later reopen drops whole, losing both
            // records. Mirror car-eventlog: a complete unterminated record
            // is replayed and terminated; a torn prefix is skipped and
            // truncated before any resumed append (Parslee-ai/car#1140).
            let terminated = buf.ends_with(b"\n");
            // Parse from the raw bytes, not a lossy string: `from_utf8_lossy`
            // would turn invalid UTF-8 into U+FFFD and let a corrupted record
            // parse — and mutate state with mangled values — where the
            // previous `lines()`-based reader (and `from_slice` here)
            // rejects it as malformed.
            if buf.iter().all(|byte| byte.is_ascii_whitespace()) {
                if !terminated {
                    repair = TailRepair::TruncateTo(line_start);
                    break;
                }
                continue;
            }
            if let Ok(t) = serde_json::from_slice::<StateTransition>(&buf) {
                replay_transition(&mut state, &mut transitions, &mut versions, now, t);
                if !terminated {
                    repair = TailRepair::Terminate;
                    break;
                }
                continue;
            }
            if let Ok(record) = serde_json::from_slice::<BatchTransitionRecord>(&buf) {
                // A parsed batch line is complete by construction — a torn
                // append is malformed JSON and falls through to the skip
                // below, so replay never applies a prefix of a batch
                // (Parslee-ai/car#1140).
                for t in record.batch {
                    replay_transition(&mut state, &mut transitions, &mut versions, now, t);
                }
                if !terminated {
                    repair = TailRepair::Terminate;
                    break;
                }
                continue;
            }
            if !terminated {
                repair = TailRepair::TruncateTo(line_start);
                break;
            }
            // Malformed line. Don't refuse to boot over it.
            tracing::warn!(
                journal = %path.display(),
                "skipping malformed StateStore journal line"
            );
        }
        Ok(repair)
    }

    fn append_journal(&self, transition: &StateTransition) {
        let Ok(json) = serde_json::to_string(transition) else {
            return;
        };
        self.append_journal_line(json);
    }

    /// Append one batch as a SINGLE journal line (see
    /// [`BatchTransitionRecord`]). A single-entry batch keeps the legacy
    /// bare-transition line shape; only a genuinely multi-key batch takes
    /// the batch record, so journals stay readable by earlier CAR versions
    /// until a multi-key batch actually occurs.
    fn append_journal_batch(&self, batch: &[StateTransition]) {
        match batch {
            [] => {}
            [single] => self.append_journal(single),
            _ => {
                let Ok(json) = serde_json::to_string(&BatchTransitionRecordRef { batch }) else {
                    return;
                };
                self.append_journal_line(json);
            }
        }
    }

    fn append_journal_line(&self, json: String) {
        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 path = journal.path.clone();
        let writer = match journal.writer_mut() {
            Ok(writer) => writer,
            Err(e) => {
                tracing::warn!(
                    journal = %path.display(),
                    error = %e,
                    "StateStore journal writer reopen failed"
                );
                return;
            }
        };
        if let Err(e) = writeln!(writer, "{json}") {
            tracing::warn!(
                journal = %path.display(),
                error = %e,
                "StateStore journal append failed"
            );
            return;
        }
        let _ = writer.flush();
    }

    /// Atomically replace the durable journal with `state` as it existed at
    /// the rollback boundary. The replacement file is fully flushed and
    /// fsynced before rename, so a successful return means replay cannot
    /// resurrect transitions discarded from memory.
    ///
    /// Callers hold the state lock while this runs. That matches the write
    /// path's state -> journal lock order and prevents an append from landing
    /// on the old file between snapshot creation and replacement.
    fn replace_journal_with_snapshot(
        &self,
        state: &HashMap<String, Value>,
        transitions: &[StateTransition],
        versions: &HashMap<String, u64>,
    ) -> std::io::Result<RestoreDurability> {
        let restore_failures = self.restore_failures.as_ref();
        self.replace_journal_with_snapshot_and_sync(state, transitions, versions, |path| {
            if let Some(failures) = restore_failures {
                failures.check(RestoreFailurePoint::ParentDirectorySync)?;
            }
            sync_parent_directory(path)
        })
    }

    fn replace_journal_with_snapshot_and_sync<F>(
        &self,
        state: &HashMap<String, Value>,
        transitions: &[StateTransition],
        versions: &HashMap<String, u64>,
        sync_parent: F,
    ) -> std::io::Result<RestoreDurability>
    where
        F: FnOnce(&Path) -> std::io::Result<()>,
    {
        let mut journal = self.journal.lock();
        let Some(journal) = journal.as_mut() else {
            return Ok(RestoreDurability::Durable);
        };
        journal.writer_mut()?.flush()?;

        let journal_permissions = std::fs::metadata(&journal.path)?.permissions();
        let (tmp_path, tmp_file) = create_replacement_temp(&journal.path)?;
        let mut cleanup = TempFileCleanup::new(tmp_path.clone());
        let mut replacement = BufWriter::new(tmp_file);
        let latest_by_key: HashMap<&str, &StateTransition> = transitions
            .iter()
            .map(|transition| (transition.key.as_str(), transition))
            .collect();
        let mut keys: Vec<&String> = state.keys().collect();
        keys.sort();
        for key in keys {
            let previous = latest_by_key.get(key.as_str()).copied();
            let transition = StateTransition {
                key: key.clone(),
                old_value: None,
                new_value: state.get(key).cloned(),
                action_id: previous
                    .map(|transition| transition.action_id.clone())
                    .unwrap_or_else(|| "restore".to_string()),
                timestamp: previous
                    .map(|transition| transition.timestamp)
                    .unwrap_or_else(Utc::now),
                ttl_secs: previous.and_then(|transition| transition.ttl_secs),
                version: versions.get(key).copied(),
            };
            let line = serde_json::to_string(&transition)?;
            writeln!(replacement, "{line}")?;
        }
        replacement.flush()?;
        replacement.get_ref().set_permissions(journal_permissions)?;
        replacement.get_ref().sync_all()?;
        if let Some(failures) = &self.restore_failures {
            failures.check(RestoreFailurePoint::BeforePublication)?;
        }
        #[cfg(target_os = "windows")]
        {
            // MoveFileExW cannot reliably replace a destination while either
            // the old journal or replacement temp is open, even when the
            // handles share delete access. Both files are already fsynced, so
            // close them before publication and reopen the destination for
            // subsequent appends.
            drop(journal.writer.take());
            drop(replacement);
            if let Err(publication_error) = replace_file_atomically(&tmp_path, &journal.path) {
                return match journal.reopen_writer() {
                    Ok(()) => Err(publication_error),
                    Err(reopen_error) => Err(std::io::Error::other(format!(
                        "state journal replacement failed ({publication_error}); reopening the original journal also failed ({reopen_error})"
                    ))),
                };
            }
            cleanup.disarm();
            if let Err(error) = journal.reopen_writer() {
                let error = format!(
                    "replacement journal was published but its append writer could not be reopened: {error}"
                );
                journal.pending_parent_sync = Some(error.clone());
                return Ok(RestoreDurability::DurabilityUnknown { error });
            }
        }
        #[cfg(not(target_os = "windows"))]
        {
            replace_file_atomically(&tmp_path, &journal.path)?;
            cleanup.disarm();
            journal.writer = Some(replacement);
        }
        match sync_parent(&journal.path) {
            Ok(()) => {
                journal.pending_parent_sync = None;
                Ok(RestoreDurability::Durable)
            }
            Err(error) => {
                let error = error.to_string();
                journal.pending_parent_sync = Some(error.clone());
                Ok(RestoreDurability::DurabilityUnknown { error })
            }
        }
    }

    fn reconcile_pending_parent_sync(&self) -> std::io::Result<()> {
        let mut journal = self.journal.lock();
        let Some(journal) = journal.as_mut() else {
            return Ok(());
        };
        if journal.pending_parent_sync.is_none() {
            return Ok(());
        }
        let writer = journal.writer_mut()?;
        writer.flush()?;
        writer.get_ref().sync_all()?;
        if let Some(failures) = &self.restore_failures {
            failures.check(RestoreFailurePoint::ParentDirectorySync)?;
        }
        sync_parent_directory(&journal.path)?;
        journal.pending_parent_sync = None;
        Ok(())
    }

    #[cfg(test)]
    fn pause_before_mutation_state_lock(&self) {
        if let Some(barrier) = &self.mutation_before_state_lock {
            barrier.pause_mutation();
        }
    }

    fn lock_reconciled_state_for_mutation(
        &self,
    ) -> std::io::Result<MutexGuard<'_, HashMap<String, Value>>> {
        #[cfg(test)]
        self.pause_before_mutation_state_lock();
        let state = self.state.lock();
        self.reconcile_pending_parent_sync()?;
        Ok(state)
    }

    fn require_reconciled_state_for_mutation(&self) -> MutexGuard<'_, HashMap<String, Value>> {
        self.lock_reconciled_state_for_mutation()
            .unwrap_or_else(|error| {
                panic!(
                    "StateStore journal is durability-unknown; refusing mutation until parent sync succeeds: {error}"
                )
            })
    }

    /// 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(());
        };
        let writer = journal.writer_mut()?;
        writer.flush()?;
        writer.get_ref().sync_all()?;
        if journal.pending_parent_sync.is_some() {
            sync_parent_directory(&journal.path)?;
            journal.pending_parent_sync = None;
        }
        Ok(())
    }

    /// 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>> {
        self.reap_expired_where(now, |_| true)
    }

    /// Reap only the expired keys in one tenant's namespace (EPIC E / E3).
    /// `tenant = Some(id)` reaps `tenant:<id>:*`; `tenant = None` reaps only
    /// the unscoped namespace. This is the per-tenant counterpart to
    /// [`Self::reap_expired`] (which reaps across all tenants): it lets a
    /// per-tenant reaping budget expire one tenant's TTL'd keys without
    /// touching another tenant's — so one tenant's memory pressure can't
    /// evict another's state.
    pub fn reap_expired_scoped(
        &self,
        now: DateTime<Utc>,
        tenant: Option<&str>,
    ) -> std::io::Result<Vec<String>> {
        self.reap_expired_where(now, |k| key_in_tenant_namespace(k, tenant))
    }

    /// Shared reaping core: reap every expired key for which `keep` returns
    /// true. Walks the transition log once to find the latest state per key.
    fn reap_expired_where(
        &self,
        now: DateTime<Utc>,
        keep: impl Fn(&str) -> bool,
    ) -> std::io::Result<Vec<String>> {
        let mut state = self.lock_reconciled_state_for_mutation()?;
        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
            .values()
            .filter_map(|t| {
                if !keep(&t.key) {
                    return None;
                }
                let ttl = t.ttl_secs?;
                t.new_value.as_ref()?;
                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.
    ///
    /// The state lock stays held until the replacement is installed, so a
    /// concurrent mutation cannot append to the old file after the snapshot.
    pub(crate) fn compact_journal(&self) -> std::io::Result<()> {
        let state = self.state.lock();
        let transitions = self.transitions.lock();
        let versions = self.versions.lock();
        match self.replace_journal_with_snapshot(&state, &transitions, &versions)? {
            RestoreDurability::Durable => Ok(()),
            RestoreDurability::DurabilityUnknown { error } => Err(std::io::Error::other(format!(
                "state journal compaction is durability-unknown: {error}"
            ))),
        }
    }

    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.require_reconciled_state_for_mutation();
        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
    }

    /// Apply several writes as ONE atomic mutation boundary
    /// (Parslee-ai/car#1140). The state lock is held across every entry, so
    /// a concurrent reader observes the complete old state or the complete
    /// new state — never a prefix of the batch — and the journal receives
    /// the whole batch as a single line ([`BatchTransitionRecord`]), giving
    /// replay after an interrupted append the same old-or-complete
    /// guarantee.
    ///
    /// Entries apply in the order given; a duplicated key's later entry
    /// wins, each bumping the key's version. An empty batch is a no-op. A
    /// single-entry batch behaves exactly like [`Self::set`], journal line
    /// shape included.
    pub fn set_batch(
        &self,
        entries: Vec<(String, Value)>,
        action_id: &str,
    ) -> Vec<StateTransition> {
        if entries.is_empty() {
            return Vec::new();
        }
        let state = self.require_reconciled_state_for_mutation();
        self.set_batch_locked(state, entries, action_id)
    }

    fn set_batch_locked(
        &self,
        mut state: MutexGuard<'_, HashMap<String, Value>>,
        entries: Vec<(String, Value)>,
        action_id: &str,
    ) -> Vec<StateTransition> {
        let timestamp = Utc::now();
        let mut batch = Vec::with_capacity(entries.len());
        for (key, value) in entries {
            let old = state.get(&key).cloned();
            state.insert(key.clone(), value.clone());
            let version = self.bump_version(&key);
            batch.push(StateTransition {
                key,
                old_value: old,
                new_value: Some(value),
                action_id: action_id.to_string(),
                timestamp,
                ttl_secs: None,
                version: Some(version),
            });
            #[cfg(test)]
            if batch.len() == 1 {
                if let Some(barrier) = &self.batch_mid_apply {
                    barrier.pause_mutation();
                }
            }
        }
        self.transitions.lock().extend(batch.iter().cloned());
        self.append_journal_batch(&batch);
        batch
    }

    /// 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())
    }

    /// [`Self::versioned_snapshot`] restricted to the keys a proposal running
    /// under `tenant` may observe ([`key_visible_to_tenant`]): shared keys and
    /// that tenant's own, never another tenant's. Keys keep their store names.
    /// This is the snapshot pre-execution checks judge a tenant-scoped
    /// proposal against, so neither a refusal nor a conflict report can echo
    /// another tenant's key or value (Parslee-ai/car#3121).
    pub fn versioned_snapshot_visible_to(
        &self,
        tenant: Option<&str>,
    ) -> (HashMap<String, Value>, HashMap<String, u64>) {
        let state = self.state.lock();
        let versions = self.versions.lock();
        (
            state
                .iter()
                .filter(|(k, _)| key_visible_to_tenant(k, tenant))
                .map(|(k, v)| (k.clone(), v.clone()))
                .collect(),
            versions
                .iter()
                .filter(|(k, _)| key_visible_to_tenant(k, tenant))
                .map(|(k, v)| (k.clone(), *v))
                .collect(),
        )
    }

    /// A read-only [`car_ir::precondition::StateView`] over the keys visible
    /// to `tenant`, under their store names — the view a check reads a
    /// `ToolCall`'s keys through in a tenant-scoped proposal, since a tool
    /// call runs on the unscoped path but must not observe another tenant.
    pub fn visible_to<'a>(&'a self, tenant: Option<&'a str>) -> TenantVisibleView<'a> {
        TenantVisibleView {
            store: self,
            tenant,
        }
    }

    pub fn delete(&self, key: &str, action_id: &str) -> Option<StateTransition> {
        let mut state = self.require_reconciled_state_for_mutation();
        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. For durable
    /// stores the restored snapshot replaces the JSONL journal before the
    /// in-memory state is published; failures leave the current state intact.
    pub fn restore(
        &self,
        snapshot: HashMap<String, Value>,
        transition_count: usize,
    ) -> std::io::Result<RestoreDurability> {
        let restore_failures = self.restore_failures.as_ref();
        self.restore_with_parent_sync(snapshot, transition_count, |path| {
            if let Some(failures) = restore_failures {
                failures.check(RestoreFailurePoint::ParentDirectorySync)?;
            }
            sync_parent_directory(path)
        })
    }

    fn restore_with_parent_sync<F>(
        &self,
        snapshot: HashMap<String, Value>,
        transition_count: usize,
        sync_parent: F,
    ) -> std::io::Result<RestoreDurability>
    where
        F: FnOnce(&Path) -> std::io::Result<()>,
    {
        let mut state = self.state.lock();
        let mut transitions = self.transitions.lock();
        let versions = self.versions.lock();
        let mut restored_transitions = transitions.clone();
        restored_transitions.truncate(transition_count);
        let durability = self.replace_journal_with_snapshot_and_sync(
            &snapshot,
            &restored_transitions,
            &versions,
            sync_parent,
        )?;
        *state = snapshot;
        *transitions = restored_transitions;
        Ok(durability)
    }

    /// Snapshot only the keys belonging to one tenant's namespace
    /// (Parslee-ai/car#187 / EPIC E task E2). `tenant = Some(id)` captures
    /// `tenant:<id>:*`; `tenant = None` captures the unscoped (non-`tenant:`)
    /// namespace. Keys are returned in their full (prefixed) form so the
    /// result round-trips through [`Self::restore_scoped`]. This is the
    /// per-tenant counterpart to [`Self::snapshot`], which captures *all*
    /// tenants and so can't be used for a tenant-isolated rollback.
    pub fn snapshot_scoped(&self, tenant: Option<&str>) -> HashMap<String, Value> {
        let state = self.state.lock();
        state
            .iter()
            .filter(|(k, _)| key_in_tenant_namespace(k, tenant))
            .map(|(k, v)| (k.clone(), v.clone()))
            .collect()
    }

    /// Restore a single tenant's namespace from a scoped snapshot, leaving
    /// every other tenant's keys untouched (EPIC E / E2). Existing keys in
    /// the target namespace are dropped and replaced by `snapshot`; keys
    /// outside it are preserved. Fixes the cross-tenant clobber where a
    /// rollback via the unscoped [`Self::restore`] wiped concurrent
    /// tenants' state.
    ///
    /// The transition log is FILTERED, not truncated (linus review C-5):
    /// only this tenant's post-snapshot transitions are discarded.
    /// Truncating shared history dropped transitions concurrent tenants
    /// committed after `transition_count`, which both falsified the audit
    /// trail and let `reap_expired*` treat another tenant's stale TTL'd
    /// transition as latest — deleting a live key. `transition_count` is
    /// the log length captured when this tenant's snapshot was taken.
    pub fn restore_scoped(
        &self,
        tenant: Option<&str>,
        snapshot: HashMap<String, Value>,
        transition_count: usize,
    ) -> std::io::Result<RestoreDurability> {
        let mut state = self.state.lock();
        let mut transitions = self.transitions.lock();
        let versions = self.versions.lock();
        let mut restored_state = state.clone();
        restored_state.retain(|k, _| !key_in_tenant_namespace(k, tenant));
        restored_state.extend(snapshot);
        let mut restored_transitions = transitions.clone();
        if transition_count < restored_transitions.len() {
            // Keep everything up to the snapshot point; after it, keep only
            // transitions that belong to OTHER namespaces.
            let tail: Vec<StateTransition> = restored_transitions
                .drain(transition_count..)
                .filter(|transition| !key_in_tenant_namespace(&transition.key, tenant))
                .collect();
            restored_transitions.extend(tail);
        }
        let durability =
            self.replace_journal_with_snapshot(&restored_state, &restored_transitions, &versions)?;
        *state = restored_state;
        *transitions = restored_transitions;
        Ok(durability)
    }

    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>) {
        let mut state = self.require_reconciled_state_for_mutation();
        *state = 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,
        }
    }
}

/// Apply one journaled transition during replay. Last-write-wins per key;
/// TTLs that already expired are dropped at replay time so stale data never
/// surfaces on first read. The version counter restores the persisted value
/// when present (survives compaction) and counts transitions for legacy
/// pre-versioning journals, taking the max so the counter stays monotonic
/// across a mix of compacted and appended lines.
fn replay_transition(
    state: &mut HashMap<String, Value>,
    transitions: &mut Vec<StateTransition>,
    versions: &mut HashMap<String, u64>,
    now: DateTime<Utc>,
    t: StateTransition,
) {
    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);
    }
    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);
}

/// `key` as the store holds it in `tenant`'s namespace: `tenant:<id>:<key>`
/// for a non-empty tenant, `key` itself otherwise. This is the key every
/// [`ScopedStateView`] read or write goes to; a check that judges a
/// tenant-scoped action against a whole-store snapshot resolves the action's
/// keys through it (Parslee-ai/car#3121).
pub fn tenant_key(tenant: Option<&str>, key: &str) -> String {
    match tenant {
        Some(t) if !t.is_empty() => format!("tenant:{t}:{key}"),
        _ => key.to_string(),
    }
}

/// Whether a proposal running under `tenant` may observe `key`: with no
/// tenant, every key; with one, the shared (non-`tenant:`) keys and that
/// tenant's own `tenant:<id>:` keys, never another tenant's.
pub fn key_visible_to_tenant(key: &str, tenant: Option<&str>) -> bool {
    match tenant {
        Some(t) if !t.is_empty() => {
            !key.starts_with("tenant:") || key_in_tenant_namespace(key, tenant)
        }
        _ => true,
    }
}

/// See [`StateStore::visible_to`].
pub struct TenantVisibleView<'a> {
    store: &'a StateStore,
    tenant: Option<&'a str>,
}

impl car_ir::precondition::StateView for TenantVisibleView<'_> {
    fn get_value(&self, key: &str) -> Option<Value> {
        key_visible_to_tenant(key, self.tenant)
            .then(|| self.store.get(key))
            .flatten()
    }
    fn key_exists(&self, key: &str) -> bool {
        key_visible_to_tenant(key, self.tenant) && self.store.exists(key)
    }
}

/// Whether `key` belongs to the namespace identified by `tenant`.
///
/// `Some(id)` (non-empty) → keys prefixed `tenant:<id>:`. `None` or empty →
/// the unscoped namespace: every key that is NOT `tenant:`-prefixed (so an
/// unscoped snapshot/restore never touches any tenant's keys). This is the
/// predicate that makes [`StateStore::snapshot_scoped`] /
/// [`StateStore::restore_scoped`] tenant-isolated.
fn key_in_tenant_namespace(key: &str, tenant: Option<&str>) -> bool {
    match tenant {
        Some(t) if !t.is_empty() => key.starts_with(&format!("tenant:{t}:")),
        _ => !key.starts_with("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 {
        tenant_key(self.tenant, key)
    }

    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)
    }

    /// Snapshot only this tenant's namespace (EPIC E / E2) — the scoped
    /// counterpart to `StateStore::snapshot`, safe to pair with
    /// [`Self::restore`] for a tenant-isolated rollback.
    pub fn snapshot(&self) -> HashMap<String, Value> {
        self.store.snapshot_scoped(self.tenant)
    }

    /// Single-lock snapshot of this tenant's namespace with the
    /// `tenant:<id>:` prefix stripped — the map a reader surface (e.g. the
    /// `state.snapshot` RPC) should return. Taken under ONE state-lock
    /// acquisition, so it can never interleave with a concurrent mutation
    /// batch the way a `keys()`-then-`get()` loop can — the result is the
    /// complete old state or the complete new state, never a mix
    /// (Parslee-ai/car#1140). Unlike [`Self::snapshot`], the stripped keys
    /// here do NOT round-trip through [`Self::restore`].
    pub fn snapshot_stripped(&self) -> HashMap<String, Value> {
        self.store
            .snapshot_scoped(self.tenant)
            .into_iter()
            .filter_map(|(key, value)| {
                self.strip_prefix(&key)
                    .map(|stripped| (stripped.to_string(), value))
            })
            .collect()
    }

    /// Restore only this tenant's namespace from a scoped snapshot, leaving
    /// other tenants untouched (EPIC E / E2).
    pub fn restore(
        &self,
        snapshot: HashMap<String, Value>,
        transition_count: usize,
    ) -> std::io::Result<RestoreDurability> {
        self.store
            .restore_scoped(self.tenant, snapshot, transition_count)
    }

    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)
    }

    /// Batch counterpart to [`Self::set`] — every entry applies inside one
    /// atomic mutation boundary, with keys transparently prefixed into this
    /// tenant's namespace (see [`StateStore::set_batch`]).
    pub fn set_batch(
        &self,
        entries: Vec<(String, Value)>,
        action_id: &str,
    ) -> Vec<StateTransition> {
        let prefixed = entries
            .into_iter()
            .map(|(key, value)| (self.full_key(&key), value))
            .collect();
        self.store.set_batch(prefixed, action_id)
    }

    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)
    }
}

/// A scoped view answers preconditions and state dependencies from its own
/// namespace, the same keys its tenant's actions read and write.
impl car_ir::precondition::StateView for ScopedStateView<'_> {
    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).unwrap();
        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).unwrap();

        // 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 durable_restore_survives_reopen() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("state.jsonl");
        {
            let store = StateStore::durable(&path).unwrap();
            store.set("existing", serde_json::json!("old"), "setup");
            let snapshot = store.snapshot();
            let transition_count = store.transition_count();

            store.set("existing", serde_json::json!("new"), "candidate");
            store.set("created", serde_json::json!(true), "candidate");
            store.restore(snapshot, transition_count).unwrap();
            store.sync().unwrap();

            assert_eq!(store.get("existing"), Some(serde_json::json!("old")));
            assert!(!store.exists("created"));
        }

        let reopened = StateStore::durable(&path).unwrap();
        assert_eq!(reopened.get("existing"), Some(serde_json::json!("old")));
        assert!(
            !reopened.exists("created"),
            "a full rollback must not resurrect candidate state after reopen"
        );
    }

    #[test]
    fn abandoned_replacement_temps_do_not_block_restore_compaction_or_reopen() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("state.jsonl");
        {
            let store = StateStore::durable(&path).unwrap();
            store.set("existing", serde_json::json!("old"), "setup");
            store.sync().unwrap();
        }

        let abandoned_transition = StateTransition {
            key: "intruder".to_string(),
            old_value: None,
            new_value: Some(serde_json::json!("must-not-replay")),
            action_id: "abandoned-temp".to_string(),
            timestamp: Utc::now(),
            ttl_secs: None,
            version: Some(99),
        };
        let abandoned_contents = format!(
            "{}\n",
            serde_json::to_string(&abandoned_transition).unwrap()
        );
        let abandoned_temps = [
            path.with_extension("jsonl.restore.tmp"),
            dir.path().join(".state.jsonl.restore.123.456.0.tmp"),
            dir.path().join(".state.jsonl.restore.789.012.1.tmp"),
        ];
        for temp in &abandoned_temps {
            std::fs::write(temp, &abandoned_contents).unwrap();
        }

        let store = StateStore::durable(&path).unwrap();
        assert!(!store.exists("intruder"));
        let snapshot = store.snapshot();
        let transition_count = store.transition_count();
        store.set("existing", serde_json::json!("new"), "candidate");
        store.set("candidate_only", serde_json::json!(true), "candidate");
        assert_eq!(
            store.restore(snapshot, transition_count).unwrap(),
            RestoreDurability::Durable
        );
        store.set_with_ttl("expired", serde_json::json!(true), "ttl", 0);
        store.sync().unwrap();
        assert_eq!(
            store
                .reap_expired(Utc::now() + Duration::seconds(1))
                .unwrap(),
            vec!["expired".to_string()]
        );
        store.sync().unwrap();
        drop(store);

        let reopened = StateStore::durable(&path).unwrap();
        assert_eq!(reopened.get("existing"), Some(serde_json::json!("old")));
        assert!(!reopened.exists("candidate_only"));
        assert!(!reopened.exists("expired"));
        assert!(!reopened.exists("intruder"));
        for temp in &abandoned_temps {
            assert!(temp.exists(), "abandoned sibling temp should be ignored");
        }
    }

    #[test]
    fn durable_restore_parent_sync_failure_keeps_live_and_reopened_state_coherent() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("state.jsonl");
        let store = StateStore::durable(&path).unwrap();
        store.set("existing", serde_json::json!("old"), "setup");
        let snapshot = store.snapshot();
        let transition_count = store.transition_count();
        store.set("existing", serde_json::json!("new"), "candidate");

        #[cfg(unix)]
        let original_mode = {
            use std::os::unix::fs::PermissionsExt;
            std::fs::metadata(&path).unwrap().permissions().mode()
        };
        let durability = store
            .restore_with_parent_sync(snapshot, transition_count, |_| {
                Err(std::io::Error::other(
                    "injected parent directory sync failure",
                ))
            })
            .unwrap();
        assert!(matches!(
            durability,
            RestoreDurability::DurabilityUnknown { ref error }
                if error.contains("parent directory sync failure")
        ));
        assert_eq!(
            store.get("existing"),
            Some(serde_json::json!("old")),
            "once rename publishes the rollback journal, memory must adopt the same state"
        );
        assert!(
            !path.with_extension("jsonl.restore.tmp").exists(),
            "replacement temp must be cleaned after atomic publication"
        );
        #[cfg(unix)]
        {
            use std::os::unix::fs::PermissionsExt;
            assert_eq!(
                std::fs::metadata(&path).unwrap().permissions().mode(),
                original_mode,
                "replacement must preserve journal permissions"
            );
        }
        store.set("after_unknown", serde_json::json!("safe"), "later");
        store.sync().unwrap();
        drop(store);

        let reopened = StateStore::durable(&path).unwrap();
        assert_eq!(
            reopened.get("existing"),
            Some(serde_json::json!("old")),
            "the visible journal and live rollback state must agree"
        );
        assert_eq!(
            reopened.get("after_unknown"),
            Some(serde_json::json!("safe")),
            "a later write must first reconcile the pending parent sync"
        );
    }

    #[test]
    fn durability_unknown_store_refuses_write_until_parent_sync_reconciles() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("state.jsonl");
        let failures = RestoreFailureInjector::default();
        let store =
            StateStore::durable_with_restore_failure_injector(&path, Some(failures.clone()))
                .unwrap();
        store.set("existing", serde_json::json!("old"), "setup");
        let snapshot = store.snapshot();
        let transition_count = store.transition_count();
        store.set("existing", serde_json::json!("new"), "candidate");
        failures.fail_next(RestoreFailurePoint::ParentDirectorySync);
        failures.fail_next(RestoreFailurePoint::ParentDirectorySync);

        assert!(matches!(
            store.restore(snapshot, transition_count).unwrap(),
            RestoreDurability::DurabilityUnknown { .. }
        ));
        let refused = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
            store.set("must_not_write", serde_json::json!(true), "later");
        }));
        assert!(refused.is_err());
        assert!(!store.exists("must_not_write"));
        drop(store);

        let reopened = StateStore::durable(&path).unwrap();
        assert_eq!(reopened.get("existing"), Some(serde_json::json!("old")));
        assert!(!reopened.exists("must_not_write"));
    }

    #[test]
    fn mutation_waiting_for_state_reconciles_restore_durability_unknown() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("state.jsonl");
        let failures = RestoreFailureInjector::default();
        let mut store =
            StateStore::durable_with_restore_failure_injector(&path, Some(failures.clone()))
                .unwrap();
        store.set("existing", serde_json::json!("old"), "setup");
        let snapshot = store.snapshot();
        let transition_count = store.transition_count();
        store.set("existing", serde_json::json!("candidate"), "candidate");

        let barrier = Arc::new(MutationRaceBarrier::new());
        store.mutation_before_state_lock = Some(barrier.clone());
        let store = Arc::new(store);
        failures.fail_next(RestoreFailurePoint::ParentDirectorySync);
        failures.fail_next(RestoreFailurePoint::ParentDirectorySync);

        let mutation_store = store.clone();
        let mutation = std::thread::spawn(move || {
            std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
                mutation_store.set("must_not_write", serde_json::json!(true), "racing");
            }))
        });

        barrier.reached.wait();
        assert!(matches!(
            store.restore(snapshot, transition_count).unwrap(),
            RestoreDurability::DurabilityUnknown { .. }
        ));
        barrier.release.wait();

        assert!(
            mutation.join().unwrap().is_err(),
            "a mutation admitted before restore must reconcile the journal after acquiring state"
        );
        assert!(!store.exists("must_not_write"));
        drop(store);

        let reopened = StateStore::durable(&path).unwrap();
        assert_eq!(reopened.get("existing"), Some(serde_json::json!("old")));
        assert!(!reopened.exists("must_not_write"));
    }

    #[test]
    fn durable_scoped_restore_survives_reopen_without_clobbering_other_tenants() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("state.jsonl");
        {
            let store = StateStore::durable(&path).unwrap();
            store
                .scoped(Some("acme"))
                .set("existing", serde_json::json!("old"), "setup");
            store
                .scoped(Some("globex"))
                .set("survivor", serde_json::json!("before"), "setup");
            let snapshot = store.snapshot_scoped(Some("acme"));
            let transition_count = store.transition_count();

            store
                .scoped(Some("acme"))
                .set("existing", serde_json::json!("new"), "candidate");
            store
                .scoped(Some("acme"))
                .set("created", serde_json::json!(true), "candidate");
            store.scoped(Some("globex")).set(
                "survivor",
                serde_json::json!("after"),
                "other-proposal",
            );
            store
                .restore_scoped(Some("acme"), snapshot, transition_count)
                .unwrap();
            store.sync().unwrap();

            assert_eq!(
                store.scoped(Some("acme")).get("existing"),
                Some(serde_json::json!("old"))
            );
            assert!(!store.scoped(Some("acme")).exists("created"));
            assert_eq!(
                store.scoped(Some("globex")).get("survivor"),
                Some(serde_json::json!("after"))
            );
        }

        let reopened = StateStore::durable(&path).unwrap();
        assert_eq!(
            reopened.scoped(Some("acme")).get("existing"),
            Some(serde_json::json!("old"))
        );
        assert!(!reopened.scoped(Some("acme")).exists("created"));
        assert_eq!(
            reopened.scoped(Some("globex")).get("survivor"),
            Some(serde_json::json!("after")),
            "scoped rollback must preserve unrelated tenant state after reopen"
        );
    }

    #[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 scoped_reap_isolates_tenants() {
        // Each tenant has a TTL'd key that's expired. Reaping tenant "a"
        // must drop only a's key, leaving b's and the unscoped key intact —
        // one tenant's memory pressure can't evict another's (E3).
        let store = StateStore::new();
        store
            .scoped(Some("a"))
            .set_with_ttl("k", serde_json::json!(1), "set", 0);
        store
            .scoped(Some("b"))
            .set_with_ttl("k", serde_json::json!(2), "set", 0);
        store.set_with_ttl("global", serde_json::json!(3), "set", 0);

        let future = Utc::now() + Duration::seconds(10);
        let reaped = store.reap_expired_scoped(future, Some("a")).unwrap();
        assert_eq!(reaped, vec!["tenant:a:k".to_string()]);
        // Only a's key is gone.
        assert!(!store.scoped(Some("a")).exists("k"));
        assert!(store.scoped(Some("b")).exists("k"));
        assert!(store.exists("global"));

        // Reaping the unscoped namespace drops only the unscoped key.
        let reaped = store.reap_expired_scoped(future, None).unwrap();
        assert_eq!(reaped, vec!["global".to_string()]);
        assert!(store.scoped(Some("b")).exists("k"));
    }

    #[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 invalid_utf8_journal_line_is_skipped_never_applied_mangled() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("state.jsonl");
        // A record whose value bytes were corrupted on disk: everything
        // parses as JSON if the invalid byte is smoothed to U+FFFD, which
        // is exactly what a lossy reader would do — and then a corrupted
        // value would replay into state. The reader must reject the line
        // wholesale instead (matching the pre-batch `lines()` behavior).
        let mut journal = Vec::new();
        journal.extend_from_slice(
            b"{\"key\":\"good\",\"old_value\":null,\"new_value\":1,\"action_id\":\"x\",\"timestamp\":\"2026-05-11T00:00:00Z\"}\n",
        );
        journal.extend_from_slice(
            b"{\"key\":\"corrupt\",\"old_value\":null,\"new_value\":\"va\xFFlue\",\"action_id\":\"x\",\"timestamp\":\"2026-05-11T00:00:00Z\"}\n",
        );
        journal.extend_from_slice(
            b"{\"batch\":[{\"key\":\"corrupt_batch\",\"old_value\":null,\"new_value\":\"a\xFFb\",\"action_id\":\"x\",\"timestamp\":\"2026-05-11T00:00:00Z\"}]}\n",
        );
        std::fs::write(&path, &journal).unwrap();

        let store = StateStore::durable(&path).unwrap();
        assert_eq!(store.get("good"), Some(json!(1)));
        assert!(
            !store.exists("corrupt"),
            "an invalid-UTF-8 record must be skipped, not applied with U+FFFD"
        );
        assert!(!store.exists("corrupt_batch"));
    }

    #[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_restore_does_not_clobber_other_tenants() {
        // The E2 fix: a tenant's rollback must restore only its own
        // namespace, leaving concurrent tenants' state intact.
        let store = StateStore::new();
        store.scoped(Some("acme")).set("k", json!("acme-v1"), "a");
        store
            .scoped(Some("globex"))
            .set("k", json!("globex-v1"), "a");
        store.set("global", json!("g-v1"), "a");

        // Snapshot acme's namespace, then both tenants + global mutate.
        let acme_snap = store.scoped(Some("acme")).snapshot();
        store.scoped(Some("acme")).set("k", json!("acme-v2"), "a");
        store
            .scoped(Some("globex"))
            .set("k", json!("globex-v2"), "a");
        store.set("global", json!("g-v2"), "a");

        // Roll acme back. Only acme reverts; globex + global keep v2.
        store.scoped(Some("acme")).restore(acme_snap, 0).unwrap();
        assert_eq!(store.scoped(Some("acme")).get("k"), Some(json!("acme-v1")));
        assert_eq!(
            store.scoped(Some("globex")).get("k"),
            Some(json!("globex-v2"))
        );
        assert_eq!(store.get("global"), Some(json!("g-v2")));
    }

    #[test]
    fn snapshot_scoped_captures_only_its_namespace() {
        let store = StateStore::new();
        store.set("global", json!(1), "a");
        store.scoped(Some("acme")).set("x", json!(2), "a");
        store.scoped(Some("globex")).set("y", json!(3), "a");

        let acme = store.snapshot_scoped(Some("acme"));
        assert_eq!(acme.len(), 1);
        assert!(acme.contains_key("tenant:acme:x"));

        let global = store.snapshot_scoped(None);
        assert_eq!(global.len(), 1);
        assert!(global.contains_key("global"));
    }

    #[test]
    fn unscoped_restore_leaves_tenant_keys_intact() {
        // The global (None) namespace restore must not wipe tenant keys.
        let store = StateStore::new();
        store.set("g", json!("v1"), "a");
        store.scoped(Some("acme")).set("k", json!("acme"), "a");

        let snap = store.snapshot_scoped(None);
        store.set("g", json!("v2"), "a");
        store.restore_scoped(None, snap, 0).unwrap();

        assert_eq!(store.get("g"), Some(json!("v1")));
        // The tenant key survived the global rollback.
        assert_eq!(store.scoped(Some("acme")).get("k"), Some(json!("acme")));
    }

    #[test]
    fn scoped_restore_preserves_other_tenants_transitions() {
        // C-5 regression: a tenant-scoped rollback must FILTER the shared
        // transition log, not truncate it — truncation dropped concurrent
        // tenants' post-snapshot transitions, falsifying history and
        // letting the reaper act on a stale "latest" transition.
        let store = StateStore::new();
        store.scoped(Some("acme")).set("k", json!("a1"), "act");

        // acme snapshots here.
        let snap = store.snapshot_scoped(Some("acme"));
        let count = store.transition_count();

        // Concurrent activity after the snapshot: acme mutates (to be
        // rolled back) and globex commits (must survive).
        store.scoped(Some("acme")).set("k", json!("a2"), "act");
        store.scoped(Some("globex")).set("g", json!("gv"), "act");

        store.restore_scoped(Some("acme"), snap, count).unwrap();

        // acme's post-snapshot transition is gone; globex's survived.
        let tail = store.transitions_since(count);
        assert_eq!(
            tail.len(),
            1,
            "exactly globex's transition survives: {tail:?}"
        );
        assert_eq!(tail[0].key, "tenant:globex:g");
        // Values match: acme rolled back, globex untouched.
        assert_eq!(store.scoped(Some("acme")).get("k"), Some(json!("a1")));
        assert_eq!(store.scoped(Some("globex")).get("g"), Some(json!("gv")));
    }

    #[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)));
    }

    // Atomic batch mutations — Parslee-ai/car#1140.

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

        let batch = store.set_batch(
            vec![("a".to_string(), json!("new")), ("b".to_string(), json!(2))],
            "callback-action",
        );

        assert_eq!(store.get("a"), Some(json!("new")));
        assert_eq!(store.get("b"), Some(json!(2)));
        assert_eq!(store.version("a"), Some(2));
        assert_eq!(store.version("b"), Some(1));
        // Returned transitions carry the pre-batch old values and the
        // shared attribution.
        assert_eq!(batch.len(), 2);
        assert_eq!(batch[0].old_value, Some(json!("old")));
        assert_eq!(batch[1].old_value, None);
        assert!(batch.iter().all(|t| t.action_id == "callback-action"));
        // The audit log gained one transition per key.
        assert_eq!(store.transition_count(), 3);
    }

    #[test]
    fn empty_batch_is_a_noop() {
        let store = StateStore::new();
        assert!(store.set_batch(Vec::new(), "noop").is_empty());
        assert_eq!(store.transition_count(), 0);
    }

    #[test]
    fn durable_two_and_three_key_batches_survive_reopen_with_versions() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("state.jsonl");
        {
            let store = StateStore::durable(&path).unwrap();
            store.set_batch(
                vec![("x".to_string(), json!(1)), ("y".to_string(), json!(2))],
                "two-key",
            );
            store.set_batch(
                vec![
                    ("x".to_string(), json!(10)),
                    ("y".to_string(), json!(20)),
                    ("z".to_string(), json!(30)),
                ],
                "three-key",
            );
            store.sync().unwrap();
        }
        let reopened = StateStore::durable(&path).unwrap();
        assert_eq!(reopened.get("x"), Some(json!(10)));
        assert_eq!(reopened.get("y"), Some(json!(20)));
        assert_eq!(reopened.get("z"), Some(json!(30)));
        assert_eq!(reopened.version("x"), Some(2));
        assert_eq!(reopened.version("y"), Some(2));
        assert_eq!(reopened.version("z"), Some(1));
    }

    #[test]
    fn multi_key_batch_is_one_journal_line_and_single_key_keeps_legacy_shape() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("state.jsonl");
        let store = StateStore::durable(&path).unwrap();

        store.set_batch(
            vec![
                ("k1".to_string(), json!(1)),
                ("k2".to_string(), json!(2)),
                ("k3".to_string(), json!(3)),
            ],
            "multi",
        );
        store.set_batch(vec![("solo".to_string(), json!(4))], "single");
        store.sync().unwrap();

        let journal = std::fs::read_to_string(&path).unwrap();
        let lines: Vec<&str> = journal.lines().filter(|l| !l.trim().is_empty()).collect();
        assert_eq!(
            lines.len(),
            2,
            "a 3-key batch must be ONE line and a 1-key batch one line: {journal}"
        );
        let multi: Value = serde_json::from_str(lines[0]).unwrap();
        assert_eq!(
            multi["batch"].as_array().map(Vec::len),
            Some(3),
            "the multi-key line is a batch record"
        );
        // The single-entry batch keeps the legacy bare-transition shape so
        // earlier CAR versions can still read journals with no multi-key
        // batches in them.
        let single: StateTransition = serde_json::from_str(lines[1]).unwrap();
        assert_eq!(single.key, "solo");
    }

    /// Shared torn-append scenario for the deterministic failure matrix:
    /// a base key, a complete batch, then an `n_keys` batch whose journal
    /// append was interrupted mid-line. Replay must surface the complete
    /// old state — never any prefix of the torn batch.
    fn torn_batch_replays_complete_old_state(n_keys: usize) {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("state.jsonl");
        let torn_keys: Vec<String> = (1..=n_keys).map(|i| format!("t{i}")).collect();
        {
            let store = StateStore::durable(&path).unwrap();
            store.set("base", json!("kept"), "setup");
            store.set_batch(
                vec![("p".to_string(), json!(1)), ("q".to_string(), json!(2))],
                "complete-batch",
            );
            store.set_batch(
                torn_keys
                    .iter()
                    .map(|key| (key.clone(), json!(format!("{key}-value"))))
                    .collect(),
                "torn-batch",
            );
            store.sync().unwrap();
        }
        let full = std::fs::read_to_string(&path).unwrap();
        let last_line_start = full.trim_end().rfind('\n').unwrap() + 1;
        let torn_cut = last_line_start + (full.trim_end().len() - last_line_start) / 2;
        std::fs::write(&path, &full.as_bytes()[..torn_cut]).unwrap();

        let store = StateStore::durable(&path).unwrap();
        assert_eq!(store.get("base"), Some(json!("kept")));
        assert_eq!(store.get("p"), Some(json!(1)));
        assert_eq!(store.get("q"), Some(json!(2)));
        for torn_key in &torn_keys {
            assert!(
                !store.exists(torn_key),
                "an interrupted {n_keys}-key batch append must replay as \
                 complete-old — no key of the torn batch may surface ({torn_key} did)"
            );
        }
    }

    #[test]
    fn torn_two_key_batch_journal_line_replays_complete_old_state_never_a_prefix() {
        torn_batch_replays_complete_old_state(2);
    }

    #[test]
    fn torn_three_key_batch_journal_line_replays_complete_old_state_never_a_prefix() {
        torn_batch_replays_complete_old_state(3);
    }

    #[test]
    fn valid_unterminated_tail_is_terminated_so_the_next_append_survives_reopen() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("state.jsonl");
        {
            let store = StateStore::durable(&path).unwrap();
            store.set("base", json!("kept"), "setup");
            store.set_batch(
                vec![("b1".to_string(), json!(1)), ("b2".to_string(), json!(2))],
                "tail-batch",
            );
            store.sync().unwrap();
        }
        // Crash shape: the final record was fully written but its trailing
        // newline never reached disk.
        let full = std::fs::read_to_string(&path).unwrap();
        std::fs::write(&path, full.trim_end().as_bytes()).unwrap();

        {
            let store = StateStore::durable(&path).unwrap();
            assert_eq!(store.get("b1"), Some(json!(1)), "the complete tail replays");
            store.set("after", json!("survives"), "later");
            store.sync().unwrap();
        }
        // Without tail repair the post-reopen append glues onto the
        // unterminated tail, and THIS reopen drops both records as one
        // malformed merged line.
        let reopened = StateStore::durable(&path).unwrap();
        assert_eq!(reopened.get("base"), Some(json!("kept")));
        assert_eq!(reopened.get("b1"), Some(json!(1)));
        assert_eq!(reopened.get("b2"), Some(json!(2)));
        assert_eq!(reopened.get("after"), Some(json!("survives")));
    }

    #[test]
    fn torn_unterminated_tail_is_truncated_so_the_next_append_is_not_merged() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("state.jsonl");
        {
            let store = StateStore::durable(&path).unwrap();
            store.set("base", json!("kept"), "setup");
            store.set_batch(
                vec![
                    ("t1".to_string(), json!(1)),
                    ("t2".to_string(), json!(2)),
                    ("t3".to_string(), json!(3)),
                ],
                "torn-batch",
            );
            store.sync().unwrap();
        }
        // Crash shape: the batch append was interrupted mid-record.
        let full = std::fs::read_to_string(&path).unwrap();
        let last_line_start = full.trim_end().rfind('\n').unwrap() + 1;
        let torn_cut = last_line_start + (full.trim_end().len() - last_line_start) / 2;
        std::fs::write(&path, &full.as_bytes()[..torn_cut]).unwrap();

        {
            let store = StateStore::durable(&path).unwrap();
            assert!(
                !store.exists("t1"),
                "the torn batch replays as complete-old"
            );
            store.set("after", json!("survives"), "later");
            store.sync().unwrap();
        }
        let reopened = StateStore::durable(&path).unwrap();
        assert_eq!(reopened.get("base"), Some(json!("kept")));
        assert_eq!(
            reopened.get("after"),
            Some(json!("survives")),
            "the record appended after a torn tail must not merge into it and be lost"
        );
        for torn_key in ["t1", "t2", "t3"] {
            assert!(!reopened.exists(torn_key));
        }
    }

    /// Shared live-read scenario for the deterministic failure matrix. The
    /// `batch_mid_apply` seam pauses the batch after its FIRST key with the
    /// state lock still held; at that provably-partial instant the state
    /// lock must be unacquirable (`try_lock` fails — the deterministic
    /// core: no reader can enter mid-batch), and a reader admitted
    /// afterwards must observe the complete new state.
    fn reader_cannot_enter_mid_batch(n_keys: usize) {
        let keys: Vec<String> = (1..=n_keys).map(|i| format!("k{i}")).collect();
        let mut store = StateStore::new();
        for key in &keys {
            store.set(key, json!("old"), "setup");
        }
        let barrier = Arc::new(MutationRaceBarrier::new());
        store.batch_mid_apply = Some(barrier.clone());
        let store = Arc::new(store);

        let batch_store = store.clone();
        let batch_keys = keys.clone();
        let batch = std::thread::spawn(move || {
            batch_store.set_batch(
                batch_keys
                    .into_iter()
                    .map(|key| (key, json!("new")))
                    .collect(),
                "batch",
            );
        });

        // Deterministic partial-state instant: exactly one key applied,
        // n_keys - 1 still old, and the state lock held by the batch.
        barrier.reached.wait();
        assert!(
            store.state.try_lock().is_none(),
            "the state lock must be held for the whole batch — a reader \
             admitted here would observe a partial {n_keys}-key set"
        );
        let reader_store = store.clone();
        let reader_keys = keys.clone();
        let reader = std::thread::spawn(move || {
            reader_keys
                .iter()
                .rev()
                .map(|key| reader_store.get(key))
                .collect::<Vec<_>>()
        });
        barrier.release.wait();
        batch.join().unwrap();

        let observed = reader.join().unwrap();
        assert!(
            observed.iter().all(|value| value == &Some(json!("new"))),
            "a reader admitted during a batch must see the complete new \
             state, got {observed:?}"
        );
    }

    #[test]
    fn reader_cannot_enter_mid_two_key_batch_and_observes_complete_state() {
        reader_cannot_enter_mid_batch(2);
    }

    #[test]
    fn reader_cannot_enter_mid_three_key_batch_and_observes_complete_state() {
        reader_cannot_enter_mid_batch(3);
    }

    #[test]
    fn snapshot_stripped_is_single_lock_and_strips_tenant_prefixes() {
        // The reader map the state.snapshot RPC returns: built under one
        // state-lock acquisition (the same lock the mid-batch tests above
        // prove is held for a whole batch), with tenant prefixes stripped.
        let store = StateStore::new();
        store.set("global", json!(1), "act");
        store.scoped(Some("acme")).set("x", json!(2), "act");
        store.scoped(Some("globex")).set("y", json!(3), "act");

        let acme = store.scoped(Some("acme")).snapshot_stripped();
        assert_eq!(acme, [("x".to_string(), json!(2))].into());

        let unscoped = store.scoped(None).snapshot_stripped();
        assert_eq!(unscoped, [("global".to_string(), json!(1))].into());
    }

    #[test]
    fn a_tenant_sees_shared_keys_and_its_own_never_another_tenants() {
        use car_ir::precondition::StateView;
        let store = StateStore::new();
        store.set("global", json!(1), "act");
        store.scoped(Some("acme")).set("x", json!(2), "act");
        store.scoped(Some("globex")).set("y", json!(3), "act");

        let (values, versions) = store.versioned_snapshot_visible_to(Some("acme"));
        let expected: HashMap<String, Value> = [
            ("global".to_string(), json!(1)),
            ("tenant:acme:x".to_string(), json!(2)),
        ]
        .into();
        assert_eq!(values, expected);
        assert_eq!(
            versions.keys().collect::<std::collections::HashSet<_>>(),
            expected.keys().collect()
        );

        let view = store.visible_to(Some("acme"));
        assert!(view.key_exists("global"));
        assert!(view.key_exists("tenant:acme:x"));
        assert!(!view.key_exists("tenant:globex:y"));
        assert_eq!(view.get_value("tenant:globex:y"), None);

        // No tenant: the whole store, as before.
        assert_eq!(
            store.versioned_snapshot_visible_to(None),
            store.versioned_snapshot()
        );
        assert!(store.visible_to(None).key_exists("tenant:globex:y"));
    }

    #[test]
    fn set_batch_refused_while_journal_durability_unknown() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("state.jsonl");
        let failures = RestoreFailureInjector::default();
        let store =
            StateStore::durable_with_restore_failure_injector(&path, Some(failures.clone()))
                .unwrap();
        store.set("existing", json!("old"), "setup");
        let snapshot = store.snapshot();
        let transition_count = store.transition_count();
        store.set("existing", json!("new"), "candidate");
        failures.fail_next(RestoreFailurePoint::ParentDirectorySync);
        failures.fail_next(RestoreFailurePoint::ParentDirectorySync);

        assert!(matches!(
            store.restore(snapshot, transition_count).unwrap(),
            RestoreDurability::DurabilityUnknown { .. }
        ));
        let refused = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
            store.set_batch(
                vec![
                    ("must_not_write".to_string(), json!(true)),
                    ("nor_this".to_string(), json!(true)),
                ],
                "later",
            );
        }));
        assert!(
            refused.is_err(),
            "set_batch must refuse mutation exactly like set while the journal is durability-unknown"
        );
        assert!(!store.exists("must_not_write"));
        assert!(!store.exists("nor_this"));
    }

    #[test]
    fn scoped_set_batch_prefixes_all_keys_into_tenant_namespace() {
        let store = StateStore::new();
        store.scoped(Some("acme")).set_batch(
            vec![("a".to_string(), json!(1)), ("b".to_string(), json!(2))],
            "act",
        );
        assert_eq!(store.scoped(Some("acme")).get("a"), Some(json!(1)));
        assert_eq!(store.scoped(Some("acme")).get("b"), Some(json!(2)));
        assert!(!store.scoped(Some("globex")).exists("a"));
        assert!(!store.scoped(None).exists("a"));
        assert_eq!(store.get("tenant:acme:b"), Some(json!(2)));
    }

    #[test]
    fn batch_then_compaction_and_reopen_preserve_state_and_versions() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("state.jsonl");
        {
            let store = StateStore::durable(&path).unwrap();
            store.set_batch(
                vec![("a".to_string(), json!(1)), ("b".to_string(), json!(2))],
                "batch",
            );
            // A TTL key that expires forces a journal compaction, which
            // rewrites the batch record as per-key snapshot lines.
            store.set_with_ttl("tmp", json!(true), "ttl", 0);
            store.sync().unwrap();
            store
                .reap_expired(Utc::now() + Duration::seconds(1))
                .unwrap();
            store.sync().unwrap();
        }
        let reopened = StateStore::durable(&path).unwrap();
        assert_eq!(reopened.get("a"), Some(json!(1)));
        assert_eq!(reopened.get("b"), Some(json!(2)));
        assert_eq!(reopened.version("a"), Some(1));
        assert_eq!(reopened.version("b"), Some(1));
        assert!(!reopened.exists("tmp"));
    }
}