car-verify 0.32.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 crate::dag::{ordered, transitive_ancestors};
use car_ir::{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)
    }
}

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

/// Like [`check_transaction`], but augments each action's write set with the
/// keys a **verified Code World Model predicts** it writes (Code World Models
/// "Slice 3b" — `docs/proposals/code-world-models.md`).
///
/// `check_transaction` reasons over *declared* effects (`expected_effects` /
/// `write_set`). A tool that writes a key it didn't declare produces a
/// write-write/read-write hazard that only surfaces at runtime. Once a model has
/// been verified against trajectories (Slice 1) you can predict those writes
/// (Slice 2) and fold them into pre-flight conflict detection here:
/// `predicted_writes` maps an `Action::id` to the extra keys the model predicts
/// that action writes; they are unioned into the action's write set before the
/// same conflict analysis runs. Pure — the caller produces the predictions
/// (e.g. by running the generated model), so no model/execution dependency
/// enters this crate.
pub fn check_transaction_with_predictions(
    proposal: &ActionProposal,
    current_versions: &HashMap<String, u64>,
    current_state: Option<&HashMap<String, Value>>,
    predicted_writes: &HashMap<String, Vec<String>>,
) -> TransactionReport {
    // Build a view of the proposal whose declared write_set includes the
    // predicted keys, then reuse the exact same checker — so predicted-write
    // conflicts are detected identically to declared ones, with no logic fork.
    let mut augmented = proposal.clone();
    for action in &mut augmented.actions {
        if let Some(extra) = predicted_writes.get(&action.id) {
            for key in extra {
                if !action.write_set.contains(key) {
                    action.write_set.push(key.clone());
                }
            }
        }
    }
    check_transaction(&augmented, current_versions, current_state)
}

/// 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![],
            invocation_mode: Default::default(),
            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 predicted_writes_surface_undeclared_conflict() {
        // Neither action *declares* a write to "k", so the plain checker sees no
        // race. A verified CWM predicts both actually write "k" → the prediction
        // variant flags the hazard pre-execution.
        let a = act("a", ActionType::ToolCall);
        let b = act("b", ActionType::ToolCall);
        let p = prop(vec![a, b]);

        assert!(check_transaction(&p, &HashMap::new(), None).consistent);

        let predicted: HashMap<String, Vec<String>> = [
            ("a".to_string(), vec!["k".to_string()]),
            ("b".to_string(), vec!["k".to_string()]),
        ]
        .into();
        let r = check_transaction_with_predictions(&p, &HashMap::new(), None, &predicted);
        assert!(!r.consistent);
        assert_eq!(r.conflicts_of(ConflictKind::WriteWrite).count(), 1);
    }

    #[test]
    fn no_predictions_matches_plain_check() {
        let mut a = act("a", ActionType::ToolCall);
        a.write_set = vec!["k".to_string()];
        let b = act("b", ActionType::ToolCall);
        let p = prop(vec![a, b]);
        let empty: HashMap<String, Vec<String>> = HashMap::new();
        assert_eq!(
            check_transaction_with_predictions(&p, &HashMap::new(), None, &empty).consistent,
            check_transaction(&p, &HashMap::new(), None).consistent
        );
    }

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