car-verify 0.32.0

//! 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,
    /// No anomalies — serializable.
    L4,
}

/// The result of analyzing a schedule: the achieved consistency level, whether
/// it was serializable, and every anomaly found.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct ConcurrencyReport {
    pub level: ConsistencyLevel,
    pub serializable: bool,
    pub anomalies: Vec<AnomalyFinding>,
}

/// 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: matches!(level, ConsistencyLevel::L4),
        level,
        anomalies,
    }
}

/// The achieved level is set by the most severe anomaly present, mirroring the
/// classical isolation-level total order.
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()
        }
    }

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