behavior-contracts 0.2.4

//! bc#23: bounded parallel stage execution (`run_plan_parallel`).
//!
//! Proofs:
//!  1. overlap  — siblings in one stage actually run concurrently (condvar barrier
//!     with timeout; sequential execution times out → error outcome → test failure).
//!  2. bounded  — peak in-flight equals the plan.concurrency bound, never above.
//!  3. identical — failure identity (code + message) matches run_plan regardless of
//!     completion order. (Vector-level determinism is proven by the conformance
//!     runner, which runs every plan vector through BOTH paths and fails on any
//!     divergence.)

use std::sync::{Condvar, Mutex};
use std::time::Duration;

use behavior_contracts::{
    run_plan, run_plan_parallel, ExecOutcome, ExecutionPlanSpec, OpSpec, Value,
};

fn op(id: &str, parent: Option<usize>) -> OpSpec {
    OpSpec {
        id: id.into(),
        parent,
        bind_field: None,
        relation_kind: None,
        policy: None,
    }
}

#[test]
fn siblings_overlap_with_barrier() {
    let plan = ExecutionPlanSpec {
        groups: vec![vec![0], vec![1, 2]],
        concurrency: 2,
    };
    let ops = vec![op("root", None), op("sibA", Some(0)), op("sibB", Some(0))];

    let arrived = Mutex::new(0usize);
    let cv = Condvar::new();

    let exec = |o: &OpSpec, _b: Option<&Value>| -> ExecOutcome {
        if o.parent.is_none() {
            return ExecOutcome::Ok(Value::Obj(vec![]));
        }
        let mut n = arrived.lock().unwrap();
        *n += 1;
        cv.notify_all();
        while *n < 2 {
            let (guard, timeout) = cv.wait_timeout(n, Duration::from_secs(2)).unwrap();
            n = guard;
            if timeout.timed_out() && *n < 2 {
                // sequential execution: the sibling never became in-flight
                return ExecOutcome::Error("no-overlap".into());
            }
        }
        ExecOutcome::Ok(Value::Str(format!("{}-done", o.id)))
    };

    let res = run_plan_parallel(Some(&plan), &ops, exec)
        .expect("siblings must overlap (sequential execution would fail here)");
    assert_eq!(res.executed, vec!["root", "sibA", "sibB"]);
}

#[test]
fn inflight_bounded_by_concurrency() {
    let mut ops = vec![op("root", None)];
    for id in ["sib1", "sib2", "sib3", "sib4"] {
        ops.push(op(id, Some(0)));
    }
    let plan = ExecutionPlanSpec {
        groups: vec![vec![0], vec![1, 2, 3, 4]],
        concurrency: 2,
    };

    let state = Mutex::new((0usize, 0usize)); // (inflight, peak)
    let exec = |o: &OpSpec, _b: Option<&Value>| -> ExecOutcome {
        if o.parent.is_none() {
            return ExecOutcome::Ok(Value::Obj(vec![]));
        }
        {
            let mut s = state.lock().unwrap();
            s.0 += 1;
            s.1 = s.1.max(s.0);
        }
        std::thread::sleep(Duration::from_millis(20));
        state.lock().unwrap().0 -= 1;
        ExecOutcome::Ok(Value::Str(o.id.clone()))
    };

    let res = run_plan_parallel(Some(&plan), &ops, exec).expect("run");
    assert_eq!(res.executed.len(), 5);
    let peak = state.lock().unwrap().1;
    assert_eq!(peak, 2, "peak in-flight {peak} != concurrency bound 2");

    // concurrency = 1 must not overlap at all (sequential fallback).
    *state.lock().unwrap() = (0, 0);
    let seq_plan = ExecutionPlanSpec {
        groups: plan.groups.clone(),
        concurrency: 1,
    };
    run_plan_parallel(Some(&seq_plan), &ops, exec).expect("run");
    let peak = state.lock().unwrap().1;
    assert_eq!(peak, 1, "concurrency=1 must not overlap (peak {peak})");
}

#[test]
fn failure_identity_matches_declaration_order() {
    // sibC (index 3) fails instantly; sibB (index 2) fails late. The committed
    // failure must be sibB's OP_FAILED (first failing member in declaration
    // order), byte-identical to the sequential reference.
    let ops = vec![
        op("root", None),
        op("sibA", Some(0)),
        op("sibB", Some(0)),
        op("sibC", Some(0)),
    ];
    let groups = vec![vec![0], vec![1, 2, 3]];

    let mk = |slow: bool| {
        move |o: &OpSpec, _b: Option<&Value>| -> ExecOutcome {
            match o.id.as_str() {
                "root" => ExecOutcome::Ok(Value::Obj(vec![])),
                "sibA" => {
                    if slow {
                        std::thread::sleep(Duration::from_millis(5));
                    }
                    ExecOutcome::Ok(Value::Str("a".into()))
                }
                "sibB" => {
                    if slow {
                        std::thread::sleep(Duration::from_millis(30));
                    }
                    ExecOutcome::Error("b-broke".into())
                }
                _ => ExecOutcome::Error("c-broke-first".into()),
            }
        }
    };

    let seq_err = run_plan(
        Some(&ExecutionPlanSpec {
            groups: groups.clone(),
            concurrency: 1,
        }),
        &ops,
        mk(false),
    )
    .expect_err("sequential reference must fail");
    let par_err = run_plan_parallel(
        Some(&ExecutionPlanSpec {
            groups,
            concurrency: 3,
        }),
        &ops,
        mk(true),
    )
    .expect_err("parallel run must fail");

    assert_eq!(seq_err.code, par_err.code);
    assert_eq!(seq_err.message, par_err.message); // sibB's OP_FAILED (declaration order)
}

#[test]
fn intra_stage_dependency_falls_back_to_sequential() {
    // A (§1-violating) plan grouping a child with its parent must be executed
    // sequentially, preserving sequential visibility semantics.
    let ops = vec![
        op("root", None),
        OpSpec {
            id: "child".into(),
            parent: Some(0),
            bind_field: Some("k".into()),
            relation_kind: None,
            policy: None,
        },
    ];
    let plan = ExecutionPlanSpec {
        groups: vec![vec![0, 1]],
        concurrency: 8,
    };

    let state = Mutex::new((0usize, 0usize));
    let exec = |o: &OpSpec, _b: Option<&Value>| -> ExecOutcome {
        {
            let mut s = state.lock().unwrap();
            s.0 += 1;
            s.1 = s.1.max(s.0);
        }
        std::thread::sleep(Duration::from_millis(10));
        state.lock().unwrap().0 -= 1;
        if o.parent.is_none() {
            ExecOutcome::Ok(Value::Obj(vec![("k".into(), Value::Str("v".into()))]))
        } else {
            ExecOutcome::Ok(Value::Str("child-done".into()))
        }
    };

    let res = run_plan_parallel(Some(&plan), &ops, exec).expect("run");
    let peak = state.lock().unwrap().1;
    assert_eq!(
        peak, 1,
        "intra-stage dependency must fall back to sequential"
    );
    assert_eq!(res.executed, vec!["root", "child"]);
}