car-verify 0.52.1

//! Multi-agent concurrency anomalies — detection + consistency level
//! (arXiv 2606.17182, *Verified Detection and Prevention of Concurrency
//! Anomalies in Multi-Agent Large Language Model Systems*).
//!
//! See `docs/proposals/concurrency-anomalies.md`. Multi-agent LLM systems share
//! state (memory stores, vector indices, tool registries) through long-running
//! **read → generate → write** operations: the generate phase has real latency,
//! so another op's write can land between an op's read and its own commit. The
//! paper formalizes four anomalies — structural analogues of classical database
//! isolation anomalies — and a consistency hierarchy `L0 ⊂ … ⊂ L4`.
//!
//! This module is the pure detector + classifier. It complements
//! [`crate::transaction`] (which detects *intra-proposal* races synchronously)
//! with the *inter-agent, time-extended* lens: it takes a schedule of timestamped
//! ops and returns which anomalies occurred and the consistency level the run
//! achieved. Deterministic and side-effect-free, like the rest of the crate's
//! verifiers.

use serde::{Deserialize, Serialize};
use std::collections::HashSet;

/// One agent operation: a long-running read → generate → write. `read_at` and
/// `commit_at` are logical times (any monotonic clock) bounding the generate
/// window during which another op's commit can interleave.
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
pub struct AgentOp {
    pub id: String,
    /// The agent that ran this op (informational; anomalies are cross-agent).
    #[serde(default)]
    pub agent: String,
    /// Shared-state keys read during the read phase.
    #[serde(default)]
    pub read_set: Vec<String>,
    /// Shared-state keys written at commit.
    #[serde(default)]
    pub write_set: Vec<String>,
    /// Tool-registry entries consulted during generate (a phantom-tool surface).
    #[serde(default)]
    pub tools_read: Vec<String>,
    /// Tool-registry entries this op mutates at commit (add/remove a tool).
    #[serde(default)]
    pub tools_written: Vec<String>,
    /// Causal predecessor op ids — ops that must have committed before this one.
    #[serde(default)]
    pub depends_on: Vec<String>,
    /// Logical time the read phase observed shared state.
    #[serde(default)]
    pub read_at: u64,
    /// Logical time the write phase committed.
    #[serde(default)]
    pub commit_at: u64,
}

/// The four LLM-specific concurrency anomalies the paper formalizes.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum ConcurrencyAnomaly {
    /// Lost-update analogue: an op committed a write based on a value another op
    /// overwrote between its read and its commit.
    StaleGeneration,
    /// Phantom-read analogue: the tool registry an op consulted changed during
    /// its generate window.
    PhantomTool,
    /// Causal-consistency violation: a causally-dependent op committed before the
    /// op it depends on.
    CausalCascade,
    /// Write-skew / reorder analogue: two concurrent, unordered ops wrote a
    /// shared key, so their effects land in a nondeterministic order.
    ToolEffectReorder,
}

/// A single detected anomaly, with the key/tool and ops involved.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct AnomalyFinding {
    pub anomaly: ConcurrencyAnomaly,
    /// The shared-state key or tool the anomaly is about.
    pub key: String,
    /// Op ids involved.
    pub ops: Vec<String>,
    pub explanation: String,
}

/// CAR's adaptation of the paper's consistency hierarchy `L0 ⊂ … ⊂ L4`. The
/// achieved level is set by the *most severe* anomaly present.
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum ConsistencyLevel {
    /// Causal-cascade present — causality broken (read-uncommitted analogue).
    L0,
    /// Stale-generation present — lost update (read-committed analogue).
    L1,
    /// Phantom-tool present — unstable tool set (repeatable-read analogue).
    L2,
    /// Tool-effect reordering present — write skew (snapshot analogue).
    L3,
    /// None of the four named anomalies present. Note this is a statement
    /// about the *pattern set*, not about serializability — check
    /// [`ConcurrencyReport::serializable`] for that.
    L4,
}

/// The result of analyzing a schedule: the achieved consistency level, whether
/// any anomaly was found, and which.
///
/// `serializable` is a real conflict-serializability decision, computed by
/// [`is_conflict_serializable`]: the schedule's serialization graph is built
/// from write-write, write-read and anti-dependency conflicts (plus declared
/// `depends_on` causality) and tested for a cycle. It is **independent of**
/// `level`, which reports the most severe *named* anomaly.
///
/// The two can disagree, and that is the point. Write skew — concurrent ops
/// reading overlapping state and writing disjoint keys — matches no named
/// pattern, because [`ConcurrencyAnomaly::ToolEffectReorder`] requires the write
/// sets to intersect. Such a schedule reports `level: L4` and
/// `serializable: false`. Read `level` for "which known anomaly do I remediate",
/// and `serializable` for "does an equivalent serial order exist" (car#768).
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct ConcurrencyReport {
    pub level: ConsistencyLevel,
    pub serializable: bool,
    pub anomalies: Vec<AnomalyFinding>,
}

impl ConcurrencyReport {
    /// What kind of check produced this report — see [`crate::EvidenceTier`].
    ///
    /// Both halves are decided over the supplied schedule. Each anomaly is a
    /// structural predicate — window overlap, write-set intersection,
    /// transitive causal ancestry — evaluated exactly, and `serializable` is
    /// cycle detection over the serialization graph, which decides
    /// conflict-serializability outright.
    ///
    /// The distinction the module already draws survives intact and is worth
    /// repeating here, because a tier is easy to over-read: the four anomaly
    /// *patterns* are an incomplete set (write skew matches none of them), so
    /// an empty `anomalies` is decided but not exhaustive. That is why
    /// `serializable` exists. Incompleteness is not approximation — nothing
    /// here guesses — so the tier stays `DecisionProcedure` and the coverage
    /// caveat stays where it belongs, in the field docs above.
    pub const fn evidence_tier(&self) -> crate::EvidenceTier {
        crate::EvidenceTier::DecisionProcedure
    }
}

/// Does `a`'s generate window `[read_at, commit_at]` overlap `b`'s? Used for the
/// reorder check (concurrent windows).
fn windows_overlap(a: &AgentOp, b: &AgentOp) -> bool {
    a.read_at <= b.commit_at && b.read_at <= a.commit_at
}

/// Transitive causal ancestors of each op (by `depends_on`, resolved by id).
/// `ancestors[i]` is every op id that must causally precede op `i`.
fn causal_ancestors(ops: &[AgentOp]) -> Vec<HashSet<String>> {
    use std::collections::HashMap;
    let index: HashMap<&str, usize> = ops
        .iter()
        .enumerate()
        .map(|(i, o)| (o.id.as_str(), i))
        .collect();
    let n = ops.len();
    let mut ancestors: Vec<HashSet<String>> = vec![HashSet::new(); n];
    // Iterate to a fixpoint — depends_on edges may point in any order.
    let mut changed = true;
    while changed {
        changed = false;
        for i in 0..n {
            for dep in &ops[i].depends_on {
                if ancestors[i].insert(dep.clone()) {
                    changed = true;
                }
                if let Some(&di) = index.get(dep.as_str()) {
                    let dep_anc: Vec<String> = ancestors[di].iter().cloned().collect();
                    for a in dep_anc {
                        if ancestors[i].insert(a) {
                            changed = true;
                        }
                    }
                }
            }
        }
    }
    ancestors
}

/// Analyze a schedule of multi-agent read-generate-write ops for the paper's
/// four concurrency anomalies and classify the achieved consistency level.
/// Deterministic: a given schedule always yields the same report.
pub fn analyze(ops: &[AgentOp]) -> ConcurrencyReport {
    let mut anomalies = Vec::new();
    let ancestors = causal_ancestors(ops);
    let idx_of = |id: &str| ops.iter().position(|o| o.id == id);

    for (i, o) in ops.iter().enumerate() {
        let reads: HashSet<&String> = o.read_set.iter().collect();
        let writes_i: HashSet<&String> = o.write_set.iter().collect();
        let tools_read: HashSet<&String> = o.tools_read.iter().collect();

        for (j, w) in ops.iter().enumerate() {
            if i == j {
                continue;
            }

            // --- stale-generation: O reads+writes k; W overwrites k mid-window ---
            for key in writes_i.intersection(&reads) {
                if w.write_set.contains(*key)
                    && o.read_at < w.commit_at
                    && w.commit_at < o.commit_at
                {
                    anomalies.push(AnomalyFinding {
                        anomaly: ConcurrencyAnomaly::StaleGeneration,
                        key: (*key).clone(),
                        ops: vec![o.id.clone(), w.id.clone()],
                        explanation: format!(
                            "op '{}' read '{}', generated, then committed a write based on a value \
                             op '{}' overwrote in between (lost update)",
                            o.id, key, w.id
                        ),
                    });
                }
            }

            // --- phantom-tool: O consulted tool t; W mutated t mid-window ---
            for tool in tools_read.iter() {
                if w.tools_written.contains(*tool)
                    && o.read_at < w.commit_at
                    && w.commit_at < o.commit_at
                {
                    anomalies.push(AnomalyFinding {
                        anomaly: ConcurrencyAnomaly::PhantomTool,
                        key: (*tool).clone(),
                        ops: vec![o.id.clone(), w.id.clone()],
                        explanation: format!(
                            "op '{}' consulted tool '{}' during generate, but op '{}' changed the \
                             registry entry mid-window (phantom tool)",
                            o.id, tool, w.id
                        ),
                    });
                }
            }
        }

        // --- causal-cascade: O depends on D but committed before D ---
        for dep in &o.depends_on {
            if let Some(di) = idx_of(dep) {
                if ops[di].commit_at > o.commit_at {
                    anomalies.push(AnomalyFinding {
                        anomaly: ConcurrencyAnomaly::CausalCascade,
                        key: dep.clone(),
                        ops: vec![o.id.clone(), dep.clone()],
                        explanation: format!(
                            "op '{}' causally depends on '{}' but committed before it \
                             (causality violated)",
                            o.id, dep
                        ),
                    });
                }
            }
        }
    }

    // --- tool-effect reordering: unordered concurrent writers to a shared key ---
    for i in 0..ops.len() {
        for j in (i + 1)..ops.len() {
            // Causally ordered either way → not a reorder hazard.
            if ancestors[i].contains(&ops[j].id) || ancestors[j].contains(&ops[i].id) {
                continue;
            }
            if !windows_overlap(&ops[i], &ops[j]) {
                continue;
            }
            let wi: HashSet<&String> = ops[i].write_set.iter().collect();
            for key in wi.intersection(&ops[j].write_set.iter().collect()) {
                anomalies.push(AnomalyFinding {
                    anomaly: ConcurrencyAnomaly::ToolEffectReorder,
                    key: (*key).clone(),
                    ops: vec![ops[i].id.clone(), ops[j].id.clone()],
                    explanation: format!(
                        "ops '{}' and '{}' concurrently write '{}' with no causal ordering — \
                         their effects land in a nondeterministic order (reorder)",
                        ops[i].id, ops[j].id, key
                    ),
                });
            }
        }
    }

    let level = classify(&anomalies);
    ConcurrencyReport {
        serializable: is_conflict_serializable(ops),
        level,
        anomalies,
    }
}

/// The achieved level is set by the most severe anomaly present, mirroring the
/// classical isolation-level total order.
/// Namespace tool-registry entries so they share one key space with ordinary
/// state without colliding: a state key `x` and a tool named `x` are different
/// shared objects and must not be conflated.
fn tool_key(name: &str) -> String {
    format!("tool:{name}")
}

/// Every shared object an op reads / writes, tools included.
fn read_keys(o: &AgentOp) -> Vec<String> {
    o.read_set
        .iter()
        .cloned()
        .chain(o.tools_read.iter().map(|t| tool_key(t)))
        .collect()
}

fn write_keys(o: &AgentOp) -> Vec<String> {
    o.write_set
        .iter()
        .cloned()
        .chain(o.tools_written.iter().map(|t| tool_key(t)))
        .collect()
}

/// Build the serialization (conflict) graph: an edge `u -> v` means `u` must
/// precede `v` in any equivalent serial schedule.
///
/// Edges come from the three classical conflict kinds, plus declared causality:
///
/// * **write-write** — both commit to the same key, so the earlier commit is
///   ordered first (ties broken by id, purely so the graph is deterministic).
/// * **write-read** — `b` read a key `a` wrote. If `a` committed at or before
///   `b`'s read, `b` saw `a`'s value: `a -> b`. Otherwise `b` read a value that
///   predates `a`'s write, so `b` must come first: `b -> a` (the
///   anti-dependency).
/// * **`depends_on`** — a declared causal predecessor must precede its dependent.
///
/// Read-read pairs never conflict.
fn serialization_graph(ops: &[AgentOp]) -> Vec<HashSet<usize>> {
    use std::collections::HashMap;
    let n = ops.len();
    let mut adj: Vec<HashSet<usize>> = vec![HashSet::new(); n];
    let index: HashMap<&str, usize> = ops
        .iter()
        .enumerate()
        .map(|(i, o)| (o.id.as_str(), i))
        .collect();

    let reads: Vec<HashSet<String>> = ops
        .iter()
        .map(|o| read_keys(o).into_iter().collect())
        .collect();
    let writes: Vec<HashSet<String>> = ops
        .iter()
        .map(|o| write_keys(o).into_iter().collect())
        .collect();

    for i in 0..n {
        // Declared causality is an ordering constraint in its own right.
        for dep in &ops[i].depends_on {
            if let Some(&d) = index.get(dep.as_str()) {
                if d != i {
                    adj[d].insert(i);
                }
            }
        }
        for j in 0..n {
            if i == j {
                continue;
            }
            // write-write on any shared key: order by commit time. Only the
            // existence of a shared written key matters, not how many.
            if writes[i].intersection(&writes[j]).next().is_some() {
                let (earlier, later) = match ops[i].commit_at.cmp(&ops[j].commit_at) {
                    std::cmp::Ordering::Less => (i, j),
                    std::cmp::Ordering::Greater => (j, i),
                    // Ties broken by id purely so the graph is deterministic.
                    std::cmp::Ordering::Equal => {
                        if ops[i].id <= ops[j].id {
                            (i, j)
                        } else {
                            (j, i)
                        }
                    }
                };
                adj[earlier].insert(later);
            }
            // i writes, j reads
            if writes[i].intersection(&reads[j]).next().is_some() {
                if ops[i].commit_at <= ops[j].read_at {
                    adj[i].insert(j); // j read i's committed value
                } else {
                    adj[j].insert(i); // j read state predating i's write
                }
            }
        }
    }
    adj
}

/// Depth-first cycle detection over the serialization graph. A cycle means no
/// equivalent serial order exists — the schedule is not conflict-serializable.
fn has_cycle(adj: &[HashSet<usize>]) -> bool {
    #[derive(Clone, Copy, PartialEq)]
    enum Mark {
        Unvisited,
        InStack,
        Done,
    }
    let n = adj.len();
    let mut mark = vec![Mark::Unvisited; n];
    // Iterative DFS so a long schedule cannot blow the stack.
    for start in 0..n {
        if mark[start] != Mark::Unvisited {
            continue;
        }
        let mut stack: Vec<(usize, Vec<usize>)> =
            vec![(start, adj[start].iter().copied().collect())];
        mark[start] = Mark::InStack;
        while let Some((node, pending)) = stack.last_mut() {
            match pending.pop() {
                Some(next) => match mark[next] {
                    Mark::InStack => return true,
                    Mark::Unvisited => {
                        mark[next] = Mark::InStack;
                        stack.push((next, adj[next].iter().copied().collect()));
                    }
                    Mark::Done => {}
                },
                None => {
                    mark[*node] = Mark::Done;
                    stack.pop();
                }
            }
        }
    }
    false
}

/// Whether `ops` admits an equivalent serial order.
///
/// This is the real property, not "no named anomaly pattern matched". Write skew
/// — concurrent ops reading overlapping state and writing **disjoint** keys — is
/// the case the pattern set structurally cannot see, because
/// [`ConcurrencyAnomaly::ToolEffectReorder`] requires the write sets to
/// intersect. The conflict graph catches it through the two anti-dependency
/// edges it creates (car#768).
pub fn is_conflict_serializable(ops: &[AgentOp]) -> bool {
    !has_cycle(&serialization_graph(ops))
}

fn classify(anomalies: &[AnomalyFinding]) -> ConsistencyLevel {
    let has = |a: ConcurrencyAnomaly| anomalies.iter().any(|f| f.anomaly == a);
    if has(ConcurrencyAnomaly::CausalCascade) {
        ConsistencyLevel::L0
    } else if has(ConcurrencyAnomaly::StaleGeneration) {
        ConsistencyLevel::L1
    } else if has(ConcurrencyAnomaly::PhantomTool) {
        ConsistencyLevel::L2
    } else if has(ConcurrencyAnomaly::ToolEffectReorder) {
        ConsistencyLevel::L3
    } else {
        ConsistencyLevel::L4
    }
}

/// The severity level a single anomaly forces — the level [`classify`] would
/// demote to if this were the only anomaly present.
fn anomaly_level(a: ConcurrencyAnomaly) -> ConsistencyLevel {
    match a {
        ConcurrencyAnomaly::CausalCascade => ConsistencyLevel::L0,
        ConcurrencyAnomaly::StaleGeneration => ConsistencyLevel::L1,
        ConcurrencyAnomaly::PhantomTool => ConsistencyLevel::L2,
        ConcurrencyAnomaly::ToolEffectReorder => ConsistencyLevel::L3,
    }
}

// === Slice 2: prevention / gating ===
//
// Map a [`ConcurrencyReport`]'s anomalies to remediation actions and a
// disposition (auto-apply / require approval / abort), gated by a policy on the
// achieved consistency level. The analogue of `infoflow::gate_flow` for tool
// safety: a pure decision core; the live executor applies the chosen actions.

/// What to do about a detected anomaly — the structural fix that restores
/// consistency for that anomaly class.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(tag = "kind", rename_all = "snake_case")]
pub enum Remediation {
    /// stale-generation: re-read the key and regenerate `op` so it commits
    /// against current state instead of the stale value.
    RereadAndRegenerate { op: String, key: String },
    /// phantom-tool: pin the tool registry for `op`'s generate window (or
    /// re-validate the tool) so the consulted set can't shift under it.
    PinToolRegistry { op: String, tool: String },
    /// causal-cascade: enforce the causal edge so `dependent` commits after
    /// `cause` (serialize / abort-and-retry the dependent).
    EnforceCausalOrder { dependent: String, cause: String },
    /// tool-effect reordering: impose a deterministic order on the unordered
    /// writers to `key`.
    SerializeWriters { ops: Vec<String>, key: String },
}

/// How a remediation should be dispatched, by severity.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum Disposition {
    /// Apply automatically — a safe, mechanical fix.
    AutoRemediate,
    /// Escalate to a human before applying (HITL).
    RequireApproval,
    /// Too severe to remediate in place — abort the run.
    Abort,
}

/// A remediation paired with its dispatch disposition.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct GatedRemediation {
    pub anomaly: ConcurrencyAnomaly,
    pub remediation: Remediation,
    pub disposition: Disposition,
}

/// Policy for [`gate_concurrency`]: two level thresholds (compared with the
/// per-anomaly severity level). Defaults: abort only on `L0` (causal-cascade),
/// escalate `L1` (stale-generation) to approval, auto-remediate the rest.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ConcurrencyGatePolicy {
    /// Anomalies whose severity level is `<=` this abort the run.
    pub abort_at_or_below: ConsistencyLevel,
    /// Anomalies whose severity level is `<=` this (and above `abort_at_or_below`)
    /// require human approval; anything above is auto-remediated.
    pub require_approval_at_or_below: ConsistencyLevel,
}

impl Default for ConcurrencyGatePolicy {
    fn default() -> Self {
        Self {
            abort_at_or_below: ConsistencyLevel::L0,
            require_approval_at_or_below: ConsistencyLevel::L1,
        }
    }
}

/// The gate decision over a report: the achieved level, whether the run must
/// abort, and a remediation (with disposition) per anomaly finding.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct ConcurrencyGate {
    /// True when the report was serializable (no anomalies, nothing to do).
    pub safe: bool,
    pub level: ConsistencyLevel,
    /// True when any finding's disposition is `Abort`.
    pub abort: bool,
    pub remediations: Vec<GatedRemediation>,
}

/// Map a finding to its structural remediation.
fn remediation_for(f: &AnomalyFinding) -> Remediation {
    match f.anomaly {
        ConcurrencyAnomaly::StaleGeneration => Remediation::RereadAndRegenerate {
            op: f.ops.first().cloned().unwrap_or_default(),
            key: f.key.clone(),
        },
        ConcurrencyAnomaly::PhantomTool => Remediation::PinToolRegistry {
            op: f.ops.first().cloned().unwrap_or_default(),
            tool: f.key.clone(),
        },
        ConcurrencyAnomaly::CausalCascade => Remediation::EnforceCausalOrder {
            dependent: f.ops.first().cloned().unwrap_or_default(),
            cause: f.ops.get(1).cloned().unwrap_or_default(),
        },
        ConcurrencyAnomaly::ToolEffectReorder => Remediation::SerializeWriters {
            ops: f.ops.clone(),
            key: f.key.clone(),
        },
    }
}

/// Gate a [`ConcurrencyReport`] into remediations under a policy. Pure: it
/// decides *what* to do and *how* to dispatch it; the executor applies the
/// actions. The analogue of `infoflow::gate_flow`.
pub fn gate_concurrency(
    report: &ConcurrencyReport,
    policy: &ConcurrencyGatePolicy,
) -> ConcurrencyGate {
    let mut remediations = Vec::new();
    let mut abort = false;

    for f in &report.anomalies {
        let sev = anomaly_level(f.anomaly);
        let disposition = if sev <= policy.abort_at_or_below {
            abort = true;
            Disposition::Abort
        } else if sev <= policy.require_approval_at_or_below {
            Disposition::RequireApproval
        } else {
            Disposition::AutoRemediate
        };
        remediations.push(GatedRemediation {
            anomaly: f.anomaly,
            remediation: remediation_for(f),
            disposition,
        });
    }

    ConcurrencyGate {
        safe: report.anomalies.is_empty(),
        level: report.level,
        abort,
        remediations,
    }
}

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

    fn op(id: &str, read_at: u64, commit_at: u64) -> AgentOp {
        AgentOp {
            id: id.to_string(),
            read_at,
            commit_at,
            ..Default::default()
        }
    }

    /// The case that motivated car#768. Write skew: two ops read the same two
    /// keys concurrently and write **disjoint** keys, so each invalidates the
    /// other's read. No named anomaly matches — `ToolEffectReorder` requires the
    /// write sets to intersect — so the pattern set reports L4. The conflict
    /// graph sees the two anti-dependency edges and the cycle they form.
    #[test]
    fn write_skew_is_not_serializable_even_though_no_anomaly_matches() {
        let mut t1 = op("t1", 0, 10);
        t1.read_set = vec!["x".into(), "y".into()];
        t1.write_set = vec!["x".into()];
        let mut t2 = op("t2", 1, 11);
        t2.read_set = vec!["x".into(), "y".into()];
        t2.write_set = vec!["y".into()];

        let report = analyze(&[t1, t2]);
        assert_eq!(
            report.level,
            ConsistencyLevel::L4,
            "write skew matches no named anomaly — that is exactly the gap"
        );
        assert!(
            !report.serializable,
            "write skew has no equivalent serial order; reporting it serializable \
             was the defect in car#768"
        );
    }

    /// A strictly serial schedule on a shared key must stay serializable —
    /// otherwise the new check would be a false-alarm generator.
    #[test]
    fn serial_schedule_on_a_shared_key_is_serializable() {
        let mut a = op("a", 0, 5);
        a.read_set = vec!["x".into()];
        a.write_set = vec!["x".into()];
        let mut b = op("b", 6, 10);
        b.read_set = vec!["x".into()];
        b.write_set = vec!["x".into()];

        assert!(analyze(&[a, b]).serializable);
    }

    /// Concurrent ops on disjoint state do not conflict at all.
    #[test]
    fn disjoint_concurrent_ops_are_serializable() {
        let mut a = op("a", 0, 10);
        a.read_set = vec!["x".into()];
        a.write_set = vec!["x".into()];
        let mut b = op("b", 0, 10);
        b.read_set = vec!["y".into()];
        b.write_set = vec!["y".into()];

        assert!(analyze(&[a, b]).serializable);
    }

    /// Tool-registry entries are shared objects too, so a skew expressed
    /// through `tools_read` / `tools_written` must also be caught. This also
    /// pins that tool keys are namespaced: a state key `t` and a tool `t` are
    /// different objects and must not be conflated into a false conflict.
    #[test]
    fn write_skew_through_the_tool_registry_is_caught() {
        let mut t1 = op("t1", 0, 10);
        t1.tools_read = vec!["alpha".into(), "beta".into()];
        t1.tools_written = vec!["alpha".into()];
        let mut t2 = op("t2", 1, 11);
        t2.tools_read = vec!["alpha".into(), "beta".into()];
        t2.tools_written = vec!["beta".into()];

        assert!(!analyze(&[t1, t2]).serializable);

        // A state key named like a tool must not collide with it.
        let mut a = op("a", 0, 10);
        a.read_set = vec!["alpha".into()];
        a.write_set = vec!["alpha".into()];
        let mut b = op("b", 0, 10);
        b.tools_read = vec!["alpha".into()];
        b.tools_written = vec!["alpha".into()];
        assert!(
            analyze(&[a, b]).serializable,
            "state key 'alpha' and tool 'alpha' are distinct shared objects"
        );
    }

    /// A declared `depends_on` that contradicts the data-conflict order is a
    /// cycle: the schedule cannot be serialized consistently with its own
    /// stated causality.
    #[test]
    fn declared_causality_contradicting_conflict_order_is_a_cycle() {
        let mut a = op("a", 0, 5);
        a.write_set = vec!["x".into()];
        let mut b = op("b", 6, 10);
        b.read_set = vec!["x".into()];
        // b reads a's committed write (a -> b), but a claims to depend on b.
        a.depends_on = vec!["b".into()];

        assert!(!analyze(&[a, b]).serializable);
    }

    /// An empty or single-op schedule is trivially serializable.
    #[test]
    fn trivial_schedules_are_serializable() {
        assert!(analyze(&[]).serializable);
        let mut a = op("a", 0, 1);
        a.read_set = vec!["x".into()];
        a.write_set = vec!["x".into()];
        assert!(analyze(&[a]).serializable);
    }

    #[test]
    fn clean_schedule_is_serializable_l4() {
        // Two ops touching different keys, sequential windows.
        let mut a = op("a", 0, 1);
        a.read_set = vec!["x".into()];
        a.write_set = vec!["x".into()];
        let mut b = op("b", 2, 3);
        b.read_set = vec!["y".into()];
        b.write_set = vec!["y".into()];
        let r = analyze(&[a, b]);
        assert!(r.serializable);
        assert_eq!(r.level, ConsistencyLevel::L4);
        assert!(r.anomalies.is_empty());
    }

    #[test]
    fn stale_generation_is_l1() {
        // a reads k at t0, b writes k at t1, a commits k at t2 → a is stale.
        let mut a = op("a", 0, 2);
        a.read_set = vec!["k".into()];
        a.write_set = vec!["k".into()];
        let mut b = op("b", 1, 1);
        b.write_set = vec!["k".into()];
        let r = analyze(&[a, b]);
        assert_eq!(r.level, ConsistencyLevel::L1);
        assert!(r
            .anomalies
            .iter()
            .any(|f| f.anomaly == ConcurrencyAnomaly::StaleGeneration));
    }

    #[test]
    fn phantom_tool_is_l2() {
        let mut a = op("a", 0, 5);
        a.tools_read = vec!["search".into()];
        let mut b = op("b", 1, 2);
        b.tools_written = vec!["search".into()];
        let r = analyze(&[a, b]);
        assert_eq!(r.level, ConsistencyLevel::L2);
        assert!(r
            .anomalies
            .iter()
            .any(|f| f.anomaly == ConcurrencyAnomaly::PhantomTool));
    }

    #[test]
    fn causal_cascade_is_l0_and_dominates() {
        // d depends on c, but d commits (1) before c (5). Also throw in a stale
        // generation to prove causal-cascade is the most severe (→ L0).
        let mut c = op("c", 0, 5);
        c.read_set = vec!["k".into()];
        c.write_set = vec!["k".into()];
        let mut d = op("d", 0, 1);
        d.depends_on = vec!["c".into()];
        let mut e = op("e", 1, 2); // writes k mid c's window → stale for c
        e.write_set = vec!["k".into()];
        let r = analyze(&[c, d, e]);
        assert_eq!(r.level, ConsistencyLevel::L0);
        assert!(r
            .anomalies
            .iter()
            .any(|f| f.anomaly == ConcurrencyAnomaly::CausalCascade));
    }

    #[test]
    fn tool_effect_reorder_is_l3() {
        // Two unordered ops with overlapping windows both write k.
        let mut a = op("a", 0, 3);
        a.write_set = vec!["k".into()];
        let mut b = op("b", 1, 4);
        b.write_set = vec!["k".into()];
        let r = analyze(&[a, b]);
        assert_eq!(r.level, ConsistencyLevel::L3);
        assert!(r
            .anomalies
            .iter()
            .any(|f| f.anomaly == ConcurrencyAnomaly::ToolEffectReorder));
    }

    #[test]
    fn causal_order_suppresses_reorder() {
        // Same two writers to k, but b causally depends on a → ordered, no race.
        let mut a = op("a", 0, 3);
        a.write_set = vec!["k".into()];
        let mut b = op("b", 1, 4);
        b.write_set = vec!["k".into()];
        b.depends_on = vec!["a".into()];
        let r = analyze(&[a, b]);
        // b commits (4) after a (3), so the causal edge holds — no cascade, and
        // the dependency suppresses the reorder hazard → serializable.
        assert!(r
            .anomalies
            .iter()
            .all(|f| f.anomaly != ConcurrencyAnomaly::ToolEffectReorder));
        assert_eq!(r.level, ConsistencyLevel::L4);
    }

    #[test]
    fn non_overlapping_writers_are_not_a_reorder() {
        let mut a = op("a", 0, 1);
        a.write_set = vec!["k".into()];
        let mut b = op("b", 2, 3); // strictly after a
        b.write_set = vec!["k".into()];
        let r = analyze(&[a, b]);
        assert_eq!(r.level, ConsistencyLevel::L4);
    }

    // --- Slice 2: gating ---

    #[test]
    fn gate_clean_report_is_safe_no_remediation() {
        let report = ConcurrencyReport {
            level: ConsistencyLevel::L4,
            serializable: true,
            anomalies: vec![],
        };
        let g = gate_concurrency(&report, &ConcurrencyGatePolicy::default());
        assert!(g.safe);
        assert!(!g.abort);
        assert!(g.remediations.is_empty());
    }

    #[test]
    fn gate_stale_requires_approval_by_default() {
        let mut a = op("a", 0, 2);
        a.read_set = vec!["k".into()];
        a.write_set = vec!["k".into()];
        let mut b = op("b", 1, 1);
        b.write_set = vec!["k".into()];
        let report = analyze(&[a, b]);
        let g = gate_concurrency(&report, &ConcurrencyGatePolicy::default());
        assert!(!g.abort);
        let r = &g.remediations[0];
        assert_eq!(r.anomaly, ConcurrencyAnomaly::StaleGeneration);
        assert_eq!(r.disposition, Disposition::RequireApproval);
        assert!(matches!(
            r.remediation,
            Remediation::RereadAndRegenerate { .. }
        ));
    }

    #[test]
    fn gate_causal_cascade_aborts() {
        let mut c = op("c", 0, 5);
        let mut d = op("d", 0, 1);
        d.depends_on = vec!["c".into()];
        c.write_set = vec!["k".into()];
        let report = analyze(&[c, d]);
        let g = gate_concurrency(&report, &ConcurrencyGatePolicy::default());
        assert!(g.abort);
        assert!(g
            .remediations
            .iter()
            .any(|r| r.disposition == Disposition::Abort
                && matches!(r.remediation, Remediation::EnforceCausalOrder { .. })));
    }

    #[test]
    fn gate_reorder_auto_remediates() {
        let mut a = op("a", 0, 3);
        a.write_set = vec!["k".into()];
        let mut b = op("b", 1, 4);
        b.write_set = vec!["k".into()];
        let report = analyze(&[a, b]);
        let g = gate_concurrency(&report, &ConcurrencyGatePolicy::default());
        assert!(!g.abort);
        let r = &g.remediations[0];
        assert_eq!(r.disposition, Disposition::AutoRemediate);
        assert!(matches!(
            r.remediation,
            Remediation::SerializeWriters { .. }
        ));
    }

    #[test]
    fn strict_policy_escalates_reorder_to_approval() {
        // Raise the approval threshold to L3 so even reorder needs approval.
        let mut a = op("a", 0, 3);
        a.write_set = vec!["k".into()];
        let mut b = op("b", 1, 4);
        b.write_set = vec!["k".into()];
        let report = analyze(&[a, b]);
        let policy = ConcurrencyGatePolicy {
            abort_at_or_below: ConsistencyLevel::L0,
            require_approval_at_or_below: ConsistencyLevel::L3,
        };
        let g = gate_concurrency(&report, &policy);
        assert_eq!(g.remediations[0].disposition, Disposition::RequireApproval);
    }
}