car-verify 0.25.0

//! Transactional conflict detection for shared program state.
//!
//! Survey "Code as Agent Harness" §4.3 + §5.2.4: scaling from one agent to
//! many turns the shared state into a *harness substrate* that planners,
//! coders, testers, and humans all read and modify. The literature's
//! central gap is that synchronizing artifacts is not enough — agents must
//! agree on *assumptions*. "Each action should declare its read set, write
//! set, assumptions, version dependencies … conflicts should be detected
//! not only at the level of file diffs, but also at the level of plans,
//! retrieved evidence, and latent requirements."
//!
//! CAR is unusually well placed to close this gap: it already has a formal
//! [`car_ir::Action`] with declared effects/dependencies, a versioned
//! [`car_state::StateStore`], and snapshot/rollback. This module adds the
//! missing piece — a checker that, given a proposal plus the current state
//! versions/values, detects three conflict classes and proposes a semantic
//! resolution for each:
//!
//! - **write-write**: two actions write the same key with no ordering
//!   dependency declared between them — a last-writer-wins race.
//! - **read-write**: one action reads a key another writes, unordered — the
//!   reader may observe the pre- or post-write value nondeterministically.
//! - **stale assumption** (belief divergence): an action planned against a
//!   key at a version/value that the shared state has since moved past.

use car_ir::{dependency_edges, Action, ActionProposal, StateAssumption};
use serde::Serialize;
use serde_json::Value;
use std::collections::{HashMap, HashSet};

/// The class of a detected conflict.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum ConflictKind {
    /// Two actions write the same key without a declared ordering.
    WriteWrite,
    /// One action reads a key another writes, unordered.
    ReadWrite,
    /// An action's assumption about shared state is stale (version or value
    /// moved since it planned) — belief divergence.
    StaleAssumption,
}

/// Value equality for assumption checks that treats JSON numbers
/// numerically — `1` (int) and `1.0` (float) are equal. Stock serde_json
/// compares `Number` by its internal int/float arm, so a value that
/// round-tripped through a system that floatifies integers would otherwise
/// spuriously read as belief divergence (neo review M3). Falls back to
/// structural equality for everything else (object key order is already
/// order-independent in serde_json `Map`).
fn values_equal(a: &Value, b: &Value) -> bool {
    match (a, b) {
        (Value::Number(x), Value::Number(y)) => match (x.as_f64(), y.as_f64()) {
            (Some(fx), Some(fy)) => fx == fy,
            _ => x == y,
        },
        _ => a == b,
    }
}

/// A single detected transactional conflict, with a semantic explanation
/// and a suggested resolution (§5.2.4 calls for "semantic merge, rollback,
/// dependency-aware locking, belief-state reconciliation, conflict
/// explanation, and re-verification after merge").
#[derive(Debug, Clone, Serialize)]
pub struct TransactionConflict {
    pub kind: ConflictKind,
    /// The state key the conflict is about.
    pub key: String,
    /// Action ids involved (writers/readers, or the assuming action).
    pub actions: Vec<String>,
    /// Why this is a conflict, in terms a human or repair agent can act on.
    pub explanation: String,
    /// Suggested resolution strategy.
    pub resolution: String,
}

/// The result of a transactional consistency check over a proposal.
#[derive(Debug, Clone, Serialize)]
pub struct TransactionReport {
    /// True when no conflicts were found.
    pub consistent: bool,
    pub conflicts: Vec<TransactionConflict>,
}

impl TransactionReport {
    pub fn conflicts_of(&self, kind: ConflictKind) -> impl Iterator<Item = &TransactionConflict> {
        self.conflicts.iter().filter(move |c| c.kind == kind)
    }
}

/// Transitive ancestors of each action in the executor's dependency graph.
/// `ancestors[i]` is every action that must complete before `i` runs.
/// Built from [`dependency_edges`] — the *same* edges [`car_ir::build_dag`]
/// sequences on — so the checker's notion of "ordered" is identical to the
/// runtime's by construction (neo review C1: a re-derivation drifts from
/// the DAG and suppresses real races / invents false ones).
fn transitive_ancestors(actions: &[Action]) -> Vec<HashSet<usize>> {
    let direct = dependency_edges(actions);
    let n = actions.len();
    let mut ancestors: Vec<HashSet<usize>> = vec![HashSet::new(); n];
    // Edges only point to lower indices (`writer_idx < i`), so a single
    // ascending pass computes the full closure: i's ancestors are its
    // direct deps plus each direct dep's already-computed ancestors.
    for i in 0..n {
        for &d in &direct[i] {
            ancestors[i].insert(d);
            let d_anc: Vec<usize> = ancestors[d].iter().copied().collect();
            ancestors[i].extend(d_anc);
        }
    }
    ancestors
}

/// Are actions `i` and `j` sequenced by a (transitive) dependency — i.e.
/// is one a DAG ancestor of the other? If so a shared-key access is an
/// ordered step, not a race.
fn ordered(i: usize, j: usize, ancestors: &[HashSet<usize>]) -> bool {
    ancestors[i].contains(&j) || ancestors[j].contains(&i)
}

/// Check a proposal for transactional conflicts against the current shared
/// state. `current_versions` maps key → version (from
/// [`car_state::StateStore::versions`]); `current_state` (optional) maps
/// key → value for value-level assumption checks. Both describe the state
/// the proposal was *not* planned against but will execute against.
pub fn check_transaction(
    proposal: &ActionProposal,
    current_versions: &HashMap<String, u64>,
    current_state: Option<&HashMap<String, Value>>,
) -> TransactionReport {
    let mut conflicts = Vec::new();
    let actions = &proposal.actions;

    // Precompute effective read/write sets once.
    let writes: Vec<HashSet<String>> = actions
        .iter()
        .map(|a| a.effective_write_set().into_iter().collect())
        .collect();
    let reads: Vec<HashSet<String>> = actions
        .iter()
        .map(|a| a.effective_read_set().into_iter().collect())
        .collect();
    let ancestors = transitive_ancestors(actions);

    // --- Intra-proposal write-write and read-write hazards ---
    for i in 0..actions.len() {
        for j in (i + 1)..actions.len() {
            if ordered(i, j, &ancestors) {
                continue; // a transitive dependency sequences these — not a race
            }
            // write-write
            for key in writes[i].intersection(&writes[j]) {
                conflicts.push(TransactionConflict {
                    kind: ConflictKind::WriteWrite,
                    key: key.clone(),
                    actions: vec![actions[i].id.clone(), actions[j].id.clone()],
                    explanation: format!(
                        "actions '{}' and '{}' both write '{}' with no ordering dependency — last-writer-wins is nondeterministic",
                        actions[i].id, actions[j].id, key
                    ),
                    resolution: "declare an ordering (add the writer's key to the other action's state_dependencies), split into distinct keys, or define a semantic merge for this key".to_string(),
                });
            }
            // read-write (both directions)
            for key in reads[i].intersection(&writes[j]) {
                conflicts.push(TransactionConflict {
                    kind: ConflictKind::ReadWrite,
                    key: key.clone(),
                    actions: vec![actions[i].id.clone(), actions[j].id.clone()],
                    explanation: format!(
                        "action '{}' reads '{}' while '{}' writes it, unordered — the read may observe the pre- or post-write value",
                        actions[i].id, key, actions[j].id
                    ),
                    resolution: format!(
                        "sequence the reader after the writer (add '{}' to '{}'.state_dependencies) if the fresh value is intended, or before it if the prior value is",
                        key, actions[i].id
                    ),
                });
            }
            for key in reads[j].intersection(&writes[i]) {
                conflicts.push(TransactionConflict {
                    kind: ConflictKind::ReadWrite,
                    key: key.clone(),
                    actions: vec![actions[j].id.clone(), actions[i].id.clone()],
                    explanation: format!(
                        "action '{}' reads '{}' while '{}' writes it, unordered — the read may observe the pre- or post-write value",
                        actions[j].id, key, actions[i].id
                    ),
                    resolution: format!(
                        "sequence the reader after the writer (add '{}' to '{}'.state_dependencies) if the fresh value is intended, or before it if the prior value is",
                        key, actions[j].id
                    ),
                });
            }
        }
    }

    // --- Stale assumptions (belief divergence vs the real shared state) ---
    for action in actions {
        for assumption in &action.assumptions {
            if let Some(conflict) =
                check_assumption(action, assumption, current_versions, current_state)
            {
                conflicts.push(conflict);
            }
        }
    }

    TransactionReport {
        consistent: conflicts.is_empty(),
        conflicts,
    }
}

/// Check one assumption against the current shared state. An assumption is
/// stale if its pinned version no longer matches the current version, or
/// its pinned value no longer matches the current value (numeric-aware).
fn check_assumption(
    action: &Action,
    assumption: &StateAssumption,
    current_versions: &HashMap<String, u64>,
    current_state: Option<&HashMap<String, Value>>,
) -> Option<TransactionConflict> {
    if let Some(read_version) = assumption.read_version {
        let current = current_versions.get(&assumption.key).copied();
        if current != Some(read_version) {
            return Some(TransactionConflict {
                kind: ConflictKind::StaleAssumption,
                key: assumption.key.clone(),
                actions: vec![action.id.clone()],
                explanation: format!(
                    "action '{}' planned against '{}' at version {} but the shared state is now at {} — the plan is based on a stale read",
                    action.id,
                    assumption.key,
                    read_version,
                    current.map(|v| v.to_string()).unwrap_or_else(|| "absent".to_string())
                ),
                resolution: "re-verify the action against current state and reconcile its belief (re-plan from the new version) before executing".to_string(),
            });
        }
    }
    if let (Some(expected), Some(state)) = (&assumption.expected_value, current_state) {
        let current = state.get(&assumption.key);
        if !current.map(|c| values_equal(c, expected)).unwrap_or(false) {
            return Some(TransactionConflict {
                kind: ConflictKind::StaleAssumption,
                key: assumption.key.clone(),
                actions: vec![action.id.clone()],
                explanation: format!(
                    "action '{}' assumes '{}' == {} but the shared state holds {} — belief divergence",
                    action.id,
                    assumption.key,
                    expected,
                    current.map(|v| v.to_string()).unwrap_or_else(|| "absent".to_string())
                ),
                resolution: "reconcile the assumption against the current value (semantic merge or re-plan); do not execute on the stale belief".to_string(),
            });
        }
    }
    None
}

#[cfg(test)]
mod tests {
    use super::*;
    use car_ir::{ActionType, FailureBehavior};

    fn act(id: &str, ty: ActionType) -> Action {
        Action {
            id: id.to_string(),
            action_type: ty,
            tool: None,
            parameters: HashMap::new(),
            preconditions: vec![],
            expected_effects: HashMap::new(),
            state_dependencies: vec![],
            read_set: vec![],
            write_set: vec![],
            assumptions: vec![],
            idempotent: false,
            max_retries: 3,
            failure_behavior: FailureBehavior::Abort,
            timeout_ms: None,
            metadata: HashMap::new(),
        }
    }

    fn prop(actions: Vec<Action>) -> ActionProposal {
        ActionProposal {
            id: "p".to_string(),
            source: "test".to_string(),
            actions,
            timestamp: chrono::Utc::now(),
            context: HashMap::new(),
        }
    }

    #[test]
    fn detects_write_write_race() {
        let mut a = act("a", ActionType::ToolCall);
        a.write_set = vec!["k".to_string()];
        let mut b = act("b", ActionType::ToolCall);
        b.write_set = vec!["k".to_string()];
        let r = check_transaction(&prop(vec![a, b]), &HashMap::new(), None);
        assert!(!r.consistent);
        assert_eq!(r.conflicts_of(ConflictKind::WriteWrite).count(), 1);
    }

    #[test]
    fn declared_ordering_suppresses_write_write() {
        // To suppress a race the actions must be ordered by the *executor*,
        // which sequences on expected_effects/StateWrite (a DAG signal).
        // The ordering must come through a *distinct* key: a writes x and k;
        // b depends on x and writes k -> DAG edge b<-a -> ordered -> the
        // shared write on k is sequenced, not a race.
        let mut a = act("a", ActionType::ToolCall);
        a.expected_effects =
            [("x".to_string(), Value::from(1)), ("k".to_string(), Value::from(1))].into();
        let mut b = act("b", ActionType::ToolCall);
        b.state_dependencies = vec!["x".to_string()];
        b.expected_effects = [("k".to_string(), Value::from(2))].into();
        let r = check_transaction(&prop(vec![a, b]), &HashMap::new(), None);
        assert!(r.consistent, "{:?}", r.conflicts);
    }

    #[test]
    fn write_set_ordering_is_not_a_dag_signal_so_race_stands() {
        // write_set alone does NOT create execution ordering, so declaring
        // a downstream dep on a write_set-only key does not suppress the
        // race — the checker must agree with the executor and still flag it
        // (neo review C1: no false-negative from a non-executor "ordering").
        let mut a = act("a", ActionType::ToolCall);
        a.write_set = vec!["k".to_string()];
        let mut b = act("b", ActionType::ToolCall);
        b.write_set = vec!["k".to_string()];
        b.state_dependencies = vec!["k".to_string()];
        let r = check_transaction(&prop(vec![a, b]), &HashMap::new(), None);
        assert_eq!(r.conflicts_of(ConflictKind::WriteWrite).count(), 1);
    }

    #[test]
    fn transitive_ordering_suppresses_distant_write_write() {
        // A -> C -> B chain via expected_effects/state_dependencies. A and
        // B both write k; they are transitively ordered, so it is NOT a
        // race (neo review C2: ordering must be transitive).
        let mut a = act("a", ActionType::ToolCall);
        a.expected_effects = [("x".to_string(), Value::from(1)), ("k".to_string(), Value::from(1))].into();
        let mut c = act("c", ActionType::ToolCall);
        c.state_dependencies = vec!["x".to_string()];
        c.expected_effects = [("y".to_string(), Value::from(1))].into();
        let mut b = act("b", ActionType::ToolCall);
        b.state_dependencies = vec!["y".to_string()];
        b.expected_effects = [("k".to_string(), Value::from(2))].into();
        let r = check_transaction(&prop(vec![a, c, b]), &HashMap::new(), None);
        assert!(r.consistent, "transitively ordered, not a race: {:?}", r.conflicts);
    }

    #[test]
    fn detects_read_write_hazard() {
        let mut reader = act("reader", ActionType::ToolCall);
        reader.read_set = vec!["k".to_string()];
        let mut writer = act("writer", ActionType::ToolCall);
        writer.write_set = vec!["k".to_string()];
        let r = check_transaction(&prop(vec![reader, writer]), &HashMap::new(), None);
        assert_eq!(r.conflicts_of(ConflictKind::ReadWrite).count(), 1);
    }

    #[test]
    fn detects_stale_version_assumption() {
        let mut a = act("a", ActionType::ToolCall);
        a.assumptions = vec![StateAssumption {
            key: "config".to_string(),
            expected_value: None,
            read_version: Some(3),
        }];
        // Shared state has moved to version 4.
        let versions = [("config".to_string(), 4u64)].into();
        let r = check_transaction(&prop(vec![a]), &versions, None);
        assert!(!r.consistent);
        assert_eq!(r.conflicts_of(ConflictKind::StaleAssumption).count(), 1);
    }

    #[test]
    fn matching_version_assumption_is_consistent() {
        let mut a = act("a", ActionType::ToolCall);
        a.assumptions = vec![StateAssumption {
            key: "config".to_string(),
            expected_value: None,
            read_version: Some(4),
        }];
        let versions = [("config".to_string(), 4u64)].into();
        let r = check_transaction(&prop(vec![a]), &versions, None);
        assert!(r.consistent);
    }

    #[test]
    fn detects_stale_value_assumption() {
        let mut a = act("a", ActionType::ToolCall);
        a.assumptions = vec![StateAssumption {
            key: "mode".to_string(),
            expected_value: Some(Value::from("draft")),
            read_version: None,
        }];
        let state = [("mode".to_string(), Value::from("published"))].into();
        let r = check_transaction(&prop(vec![a]), &HashMap::new(), Some(&state));
        assert_eq!(r.conflicts_of(ConflictKind::StaleAssumption).count(), 1);
    }

    #[test]
    fn numeric_assumption_int_float_equal() {
        // expected 1 (int) vs current 1.0 (float) must NOT flag stale.
        let mut a = act("a", ActionType::ToolCall);
        a.assumptions = vec![StateAssumption {
            key: "n".to_string(),
            expected_value: Some(Value::from(1)),
            read_version: None,
        }];
        let state = [("n".to_string(), serde_json::json!(1.0))].into();
        let r = check_transaction(&prop(vec![a]), &HashMap::new(), Some(&state));
        assert!(r.consistent, "1 == 1.0: {:?}", r.conflicts);
    }

    #[test]
    fn union_write_set_does_not_shadow_expected_effects() {
        // Explicit write_set=["a"] plus expected_effects on "b": both must
        // be in the effective write set, so a race on "b" is caught.
        let mut x = act("x", ActionType::ToolCall);
        x.write_set = vec!["a".to_string()];
        x.expected_effects = [("b".to_string(), Value::from(1))].into();
        let mut y = act("y", ActionType::ToolCall);
        y.write_set = vec!["b".to_string()];
        let r = check_transaction(&prop(vec![x, y]), &HashMap::new(), None);
        assert_eq!(r.conflicts_of(ConflictKind::WriteWrite).count(), 1);
    }

    #[test]
    fn derived_write_set_from_expected_effects() {
        // Two actions that don't set write_set explicitly still conflict
        // via the effective (derived) write set.
        let mut a = act("a", ActionType::ToolCall);
        a.expected_effects = [("k".to_string(), Value::from(1))].into();
        let mut b = act("b", ActionType::ToolCall);
        b.expected_effects = [("k".to_string(), Value::from(2))].into();
        let r = check_transaction(&prop(vec![a, b]), &HashMap::new(), None);
        assert_eq!(r.conflicts_of(ConflictKind::WriteWrite).count(), 1);
    }
}