unit 0.32.0

// vm/tests.rs — Comprehensive unit tests for the Forth VM

use super::*;
use crate::types::Cell;

// -----------------------------------------------------------------------
// Test helpers
// -----------------------------------------------------------------------

fn test_vm() -> VM {
    let mut vm = VM::new();
    vm.silent = true;
    vm.load_prelude();
    vm
}

fn eval(vm: &mut VM, input: &str) -> String {
    vm.output_buffer = Some(String::new());
    for line in input.lines() {
        vm.interpret_line(line);
    }
    vm.output_buffer.take().unwrap_or_default()
}

fn eval_top(vm: &mut VM, input: &str) -> Cell {
    eval(vm, input);
    vm.stack.last().copied().unwrap_or(0)
}

#[test]
fn test_stack_push_pop() {
    let mut vm = test_vm();
    eval(&mut vm, "1 2 3");
    assert_eq!(vm.stack, vec![1, 2, 3]);
}

#[test]
fn test_stack_dup() {
    let mut vm = test_vm();
    eval(&mut vm, "5 DUP");
    assert_eq!(vm.stack, vec![5, 5]);
}

#[test]
fn test_stack_drop() {
    let mut vm = test_vm();
    eval(&mut vm, "1 2 DROP");
    assert_eq!(vm.stack, vec![1]);
}

#[test]
fn test_stack_swap() {
    let mut vm = test_vm();
    eval(&mut vm, "1 2 SWAP");
    assert_eq!(vm.stack, vec![2, 1]);
}

#[test]
fn test_stack_over() {
    let mut vm = test_vm();
    eval(&mut vm, "1 2 OVER");
    assert_eq!(vm.stack, vec![1, 2, 1]);
}

#[test]
fn test_stack_rot() {
    let mut vm = test_vm();
    eval(&mut vm, "1 2 3 ROT");
    assert_eq!(vm.stack, vec![2, 3, 1]);
}

#[test]
fn test_stack_nip() {
    let mut vm = test_vm();
    eval(&mut vm, "1 2 NIP");
    assert_eq!(vm.stack, vec![2]);
}

#[test]
fn test_stack_tuck() {
    let mut vm = test_vm();
    eval(&mut vm, "1 2 TUCK");
    assert_eq!(vm.stack, vec![2, 1, 2]);
}

#[test]
fn test_stack_2dup() {
    let mut vm = test_vm();
    eval(&mut vm, "1 2 2DUP");
    assert_eq!(vm.stack, vec![1, 2, 1, 2]);
}

#[test]
fn test_stack_dot_s() {
    let mut vm = test_vm();
    let out = eval(&mut vm, "1 2 3 .S");
    assert!(out.contains("<3>"));
    assert!(out.contains("1 2 3"));
}

// -----------------------------------------------------------------------
// 2. Arithmetic
// -----------------------------------------------------------------------

#[test]
fn test_add() {
    let mut vm = test_vm();
    assert_eq!(eval_top(&mut vm, "2 3 +"), 5);
}

#[test]
fn test_subtract() {
    let mut vm = test_vm();
    assert_eq!(eval_top(&mut vm, "10 3 -"), 7);
}

#[test]
fn test_multiply() {
    let mut vm = test_vm();
    assert_eq!(eval_top(&mut vm, "6 7 *"), 42);
}

#[test]
fn test_divide() {
    let mut vm = test_vm();
    assert_eq!(eval_top(&mut vm, "20 4 /"), 5);
}

#[test]
fn test_modulo() {
    let mut vm = test_vm();
    assert_eq!(eval_top(&mut vm, "17 5 MOD"), 2);
}

// -----------------------------------------------------------------------
// 3. Comparison and logic
// -----------------------------------------------------------------------

#[test]
fn test_equal_true() {
    let mut vm = test_vm();
    assert_eq!(eval_top(&mut vm, "5 5 ="), -1); // true
}

#[test]
fn test_equal_false() {
    let mut vm = test_vm();
    assert_eq!(eval_top(&mut vm, "5 3 ="), 0); // false
}

#[test]
fn test_greater_true() {
    let mut vm = test_vm();
    assert_eq!(eval_top(&mut vm, "5 3 >"), -1);
}

#[test]
fn test_greater_false() {
    let mut vm = test_vm();
    assert_eq!(eval_top(&mut vm, "3 5 >"), 0);
}

#[test]
fn test_less_than() {
    let mut vm = test_vm();
    assert_eq!(eval_top(&mut vm, "3 5 <"), -1);
}

#[test]
fn test_and() {
    let mut vm = test_vm();
    assert_eq!(eval_top(&mut vm, "-1 0 AND"), 0);
}

#[test]
fn test_or() {
    let mut vm = test_vm();
    assert_eq!(eval_top(&mut vm, "0 -1 OR"), -1);
}

#[test]
fn test_not() {
    let mut vm = test_vm();
    assert_eq!(eval_top(&mut vm, "-1 NOT"), 0);
}

// -----------------------------------------------------------------------
// 4. Memory
// -----------------------------------------------------------------------

#[test]
fn test_memory_store_fetch() {
    let mut vm = test_vm();
    eval(&mut vm, "42 1000 !");
    assert_eq!(eval_top(&mut vm, "1000 @"), 42);
}

// -----------------------------------------------------------------------
// 5. Word definitions
// -----------------------------------------------------------------------

#[test]
fn test_colon_definition() {
    let mut vm = test_vm();
    eval(&mut vm, ": SQUARE DUP * ;");
    assert_eq!(eval_top(&mut vm, "7 SQUARE"), 49);
}

#[test]
fn test_nested_definitions() {
    let mut vm = test_vm();
    eval(&mut vm, ": DOUBLE 2 * ;");
    eval(&mut vm, ": QUADRUPLE DOUBLE DOUBLE ;");
    assert_eq!(eval_top(&mut vm, "3 QUADRUPLE"), 12);
}

#[test]
fn test_constant() {
    let mut vm = test_vm();
    eval(&mut vm, "42 CONSTANT ANSWER");
    assert_eq!(eval_top(&mut vm, "ANSWER"), 42);
}

#[test]
fn test_variable() {
    let mut vm = test_vm();
    eval(&mut vm, "VARIABLE X 99 X ! X @");
    assert_eq!(*vm.stack.last().unwrap(), 99);
}

#[test]
fn test_recursion_factorial() {
    let mut vm = test_vm();
    eval(
        &mut vm,
        ": FACT DUP 1 > IF DUP 1 - RECURSE * ELSE DROP 1 THEN ;",
    );
    assert_eq!(eval_top(&mut vm, "5 FACT"), 120);
    vm.stack.clear();
    assert_eq!(eval_top(&mut vm, "10 FACT"), 3628800);
}

// -----------------------------------------------------------------------
// 6. Control flow
// -----------------------------------------------------------------------

#[test]
fn test_if_then() {
    let mut vm = test_vm();
    let out = eval(&mut vm, "1 IF 42 . THEN");
    assert!(out.contains("42"));
}

#[test]
fn test_if_else_then_true() {
    let mut vm = test_vm();
    let out = eval(&mut vm, "1 IF 1 . ELSE 2 . THEN");
    assert!(out.contains("1"));
    assert!(!out.contains("2"));
}

#[test]
fn test_if_else_then_false() {
    let mut vm = test_vm();
    let out = eval(&mut vm, "0 IF 1 . ELSE 2 . THEN");
    assert!(out.contains("2"));
    assert!(!out.contains("1"));
}

#[test]
fn test_do_loop() {
    let mut vm = test_vm();
    let out = eval(&mut vm, "5 0 DO I . LOOP");
    assert!(out.contains("0"));
    assert!(out.contains("4"));
}

#[test]
fn test_do_loop_sum() {
    let mut vm = test_vm();
    eval(&mut vm, "0 10 0 DO I + LOOP");
    assert_eq!(*vm.stack.last().unwrap(), 45);
}

#[test]
fn test_begin_until() {
    let mut vm = test_vm();
    eval(&mut vm, ": CD 5 BEGIN DUP . 1 - DUP 0 = UNTIL DROP ;");
    let out = eval(&mut vm, "CD");
    assert!(out.contains("5"));
    assert!(out.contains("1"));
}

#[test]
fn test_begin_while_repeat() {
    let mut vm = test_vm();
    eval(
        &mut vm,
        ": WH 5 BEGIN DUP 0 > WHILE DUP . 1 - REPEAT DROP ;",
    );
    let out = eval(&mut vm, "WH");
    assert!(out.contains("5"));
    assert!(out.contains("1"));
}

#[test]
fn test_nested_if() {
    let mut vm = test_vm();
    eval(
        &mut vm,
        ": SIGN DUP 0 > IF 1 ELSE DUP 0 < IF -1 ELSE 0 THEN THEN NIP ;",
    );
    assert_eq!(eval_top(&mut vm, "5 SIGN"), 1);
    vm.stack.clear();
    assert_eq!(eval_top(&mut vm, "-3 SIGN"), -1);
    vm.stack.clear();
    assert_eq!(eval_top(&mut vm, "0 SIGN"), 0);
}

// -----------------------------------------------------------------------
// 7. Strings and I/O
// -----------------------------------------------------------------------

#[test]
fn test_dot_quote() {
    let mut vm = test_vm();
    eval(&mut vm, ": MSG .\" hello\" ;");
    let out = eval(&mut vm, "MSG");
    assert_eq!(out, "hello");
}

#[test]
fn test_emit() {
    let mut vm = test_vm();
    let out = eval(&mut vm, "65 EMIT");
    assert_eq!(out, "A");
}

#[test]
fn test_cr() {
    let mut vm = test_vm();
    let out = eval(&mut vm, "CR");
    assert_eq!(out, "\n");
}

#[test]
fn test_dot() {
    let mut vm = test_vm();
    let out = eval(&mut vm, "42 .");
    assert_eq!(out, "42 ");
}

#[test]
fn test_type_word() {
    let mut vm = test_vm();
    // Store "Hi" at addresses 60000-60001
    eval(&mut vm, "72 60000 ! 105 60001 !");
    let out = eval(&mut vm, "60000 2 TYPE");
    assert_eq!(out, "Hi");
}

// -----------------------------------------------------------------------
// 8. Prelude words
// -----------------------------------------------------------------------

#[test]
fn test_abs() {
    let mut vm = test_vm();
    assert_eq!(eval_top(&mut vm, "-7 ABS"), 7);
    vm.stack.clear();
    assert_eq!(eval_top(&mut vm, "7 ABS"), 7);
}

#[test]
fn test_min_max() {
    let mut vm = test_vm();
    assert_eq!(eval_top(&mut vm, "3 7 MIN"), 3);
    vm.stack.clear();
    assert_eq!(eval_top(&mut vm, "3 7 MAX"), 7);
}

#[test]
fn test_negate() {
    let mut vm = test_vm();
    assert_eq!(eval_top(&mut vm, "5 NEGATE"), -5);
}

#[test]
fn test_inc_dec() {
    let mut vm = test_vm();
    assert_eq!(eval_top(&mut vm, "5 1+"), 6);
    vm.stack.clear();
    assert_eq!(eval_top(&mut vm, "5 1-"), 4);
}

#[test]
fn test_double_halve() {
    let mut vm = test_vm();
    assert_eq!(eval_top(&mut vm, "6 2*"), 12);
    vm.stack.clear();
    assert_eq!(eval_top(&mut vm, "6 2/"), 3);
}

#[test]
fn test_zero_predicates() {
    let mut vm = test_vm();
    assert_eq!(eval_top(&mut vm, "0 0="), -1);
    vm.stack.clear();
    assert_eq!(eval_top(&mut vm, "5 0="), 0);
    vm.stack.clear();
    assert_eq!(eval_top(&mut vm, "-3 0<"), -1);
    vm.stack.clear();
    assert_eq!(eval_top(&mut vm, "3 0<"), 0);
}

#[test]
fn test_not_equal() {
    let mut vm = test_vm();
    assert_eq!(eval_top(&mut vm, "3 5 <>"), -1);
    vm.stack.clear();
    assert_eq!(eval_top(&mut vm, "5 5 <>"), 0);
}

#[test]
fn test_true_false() {
    let mut vm = test_vm();
    assert_eq!(eval_top(&mut vm, "TRUE"), -1);
    vm.stack.clear();
    assert_eq!(eval_top(&mut vm, "FALSE"), 0);
}

#[test]
fn test_prelude_loads() {
    let vm = test_vm();
    // Prelude defines these words — verify they exist in the dictionary.
    let names: Vec<&str> = vec![
        "NIP", "TUCK", "2DUP", "2DROP", "ABS", "MIN", "MAX", "NEGATE", "1+", "1-", "TRUE", "FALSE",
    ];
    for name in names {
        assert!(
            vm.find_word(name).is_some(),
            "prelude word '{}' not found",
            name
        );
    }
}

// -----------------------------------------------------------------------
// 9. Introspection
// -----------------------------------------------------------------------

#[test]
fn test_see() {
    let mut vm = test_vm();
    let out = eval(&mut vm, "SEE NIP");
    assert!(out.contains("SWAP"));
    assert!(out.contains("DROP"));
}

#[test]
fn test_words_produces_output() {
    let mut vm = test_vm();
    let out = eval(&mut vm, "WORDS");
    assert!(out.contains("DUP"));
    assert!(out.contains("DROP"));
}

// -----------------------------------------------------------------------
// 10. Sandbox execution
// -----------------------------------------------------------------------

#[test]
fn test_sandbox_captures_output() {
    let mut vm = test_vm();
    let result = vm.execute_sandbox(".\" hello sandbox\"");
    assert!(result.success);
    assert_eq!(result.output, "hello sandbox");
}

#[test]
fn test_sandbox_captures_stack() {
    let mut vm = test_vm();
    let result = vm.execute_sandbox("2 3 + 4 *");
    assert!(result.success);
    assert_eq!(result.stack_snapshot, vec![20]);
}

#[test]
fn test_sandbox_timeout() {
    let mut vm = test_vm();
    vm.execution_timeout = 1; // 1 second
    eval(&mut vm, ": INF BEGIN 0 UNTIL ;");
    let result = vm.execute_sandbox("INF");
    assert!(!result.success);
    assert!(result.error.unwrap().contains("timeout"));
}

#[test]
fn test_sandbox_isolates_stack() {
    let mut vm = test_vm();
    eval(&mut vm, "100 200 300"); // main stack
    let result = vm.execute_sandbox("1 2 3");
    assert_eq!(result.stack_snapshot, vec![1, 2, 3]);
    assert_eq!(vm.stack, vec![100, 200, 300]); // preserved
}

// -----------------------------------------------------------------------
// 11. Serialization round-trip (persist)
// -----------------------------------------------------------------------

#[test]
fn test_snapshot_roundtrip() {
    let mut vm = test_vm();
    eval(&mut vm, ": TRIPLE DUP DUP + + ;");
    // Use a VARIABLE to store a value (this allocates within `here`).
    eval(&mut vm, "VARIABLE TESTVAR 42 TESTVAR !");
    let snap = vm.make_snapshot();
    let data = crate::persist::serialize_snapshot(&snap);
    let restored = crate::persist::deserialize_snapshot(&data).unwrap();
    assert_eq!(restored.here, snap.here);
    assert_eq!(restored.dictionary.len(), snap.dictionary.len());
    // Verify the word exists.
    let found = restored.dictionary.iter().any(|e| e.name == "TRIPLE");
    assert!(found, "TRIPLE not found in restored dictionary");
}

// -----------------------------------------------------------------------
// 12. Replication package
// -----------------------------------------------------------------------

#[test]
fn test_package_build_unpack() {
    let state = vec![1, 2, 3, 4, 5]; // dummy state
    let pkg = crate::spawn::build_package(&state).unwrap();
    let (binary, state_out, prelude) = crate::spawn::unpack_package(&pkg).unwrap();
    assert!(!binary.is_empty(), "binary should not be empty");
    assert_eq!(state_out, vec![1, 2, 3, 4, 5]);
    assert!(!prelude.is_empty(), "prelude should not be empty");
}

#[test]
fn test_package_invalid_magic() {
    let bad = vec![0, 0, 0, 0, 0, 0, 0, 0, 0, 0];
    assert!(crate::spawn::unpack_package(&bad).is_err());
}

// -----------------------------------------------------------------------
// 13. Mutation
// -----------------------------------------------------------------------

#[test]
fn test_mutation_and_undo() {
    let mut vm = test_vm();
    eval(&mut vm, ": DBL 2 * ;");
    assert_eq!(eval_top(&mut vm, "5 DBL"), 10);
    vm.stack.clear();
    // Mutate
    eval(&mut vm, "MUTATE-WORD\" DBL\"");
    // It may or may not change behavior, but undo should restore.
    eval(&mut vm, "UNDO-MUTATE");
    assert_eq!(eval_top(&mut vm, "5 DBL"), 10);
}

// -----------------------------------------------------------------------
// 14. Mesh wire format (encode/decode)
// -----------------------------------------------------------------------

#[test]
fn test_serialize_state_roundtrip() {
    let mut vm = test_vm();
    eval(&mut vm, ": TEST 1 2 + ;");
    let goals = crate::goals::GoalRegistry::empty();
    let data = crate::mesh::serialize_state(&vm.dictionary, &vm.memory, vm.here, Some(&goals));
    let (dict, mem, here) = crate::mesh::deserialize_state(&data).unwrap();
    assert_eq!(here, vm.here);
    assert_eq!(dict.len(), vm.dictionary.len());
    assert_eq!(mem[0..here], vm.memory[0..here]);
}

// -----------------------------------------------------------------------
// 15. Regression tests
// -----------------------------------------------------------------------

#[test]
fn test_deep_recursion() {
    let mut vm = test_vm();
    eval(
        &mut vm,
        ": FACT DUP 1 > IF DUP 1 - RECURSE * ELSE DROP 1 THEN ;",
    );
    assert_eq!(eval_top(&mut vm, "10 FACT"), 3628800);
}

#[test]
fn test_loop_in_definition() {
    let mut vm = test_vm();
    eval(&mut vm, ": SUM 0 SWAP 0 DO I + LOOP ;");
    assert_eq!(eval_top(&mut vm, "10 SUM"), 45);
}

#[test]
fn test_interpret_mode_do_loop() {
    let mut vm = test_vm();
    let out = eval(&mut vm, "5 0 DO I . LOOP");
    assert!(out.contains("0"));
    assert!(out.contains("4"));
}

#[test]
fn test_interpret_mode_if_else() {
    let mut vm = test_vm();
    let out = eval(&mut vm, "1 IF 1 . ELSE 2 . THEN");
    assert!(out.contains("1"));
    assert!(!out.contains("2"));
}

#[test]
fn test_eval_word() {
    let mut vm = test_vm();
    let out = eval(&mut vm, "EVAL\" 2 3 + .\"");
    assert!(out.contains("5"));
}

#[test]
fn test_string_in_definition() {
    let mut vm = test_vm();
    eval(&mut vm, ": HI .\" hello world\" ;");
    let out = eval(&mut vm, "HI");
    assert_eq!(out, "hello world");
}

#[test]
fn test_nested_loops() {
    let mut vm = test_vm();
    eval(&mut vm, ": GRID 3 0 DO 2 0 DO I . J . LOOP LOOP ;");
    let out = eval(&mut vm, "GRID");
    // Should have 6 pairs of I J values
    assert!(out.contains("0 0"));
    assert!(out.contains("1 2")); // I=1, J=2
}

// -----------------------------------------------------------------------
// New VM API tests
// -----------------------------------------------------------------------

#[test]
fn test_vm_new_empty_stacks() {
    let vm = VM::new();
    assert!(vm.stack.is_empty());
    assert!(vm.rstack.is_empty());
}

#[test]
fn test_vm_eval_returns_output() {
    let mut vm = test_vm();
    let out = vm.eval("42 .");
    assert_eq!(out.trim(), "42");
}

#[test]
fn test_sexp_eval_word_prints_envelope() {
    let mut vm = test_vm();
    let out = eval(&mut vm, "SEXP-EVAL\" (+ 2 3)\"");
    assert!(
        out.contains("(result :ok 1 :value (5) :output \"\")"),
        "envelope was: {}",
        out
    );
    // Sandboxed: the live REPL stack is untouched (the 5 lives in the envelope).
    assert!(
        vm.stack.is_empty(),
        "live stack should be empty, got {:?}",
        vm.stack
    );
}

#[test]
fn test_sexp_eval_word_runtime_error_envelope() {
    let mut vm = test_vm();
    let out = eval(&mut vm, "SEXP-EVAL\" (drop)\"");
    assert!(
        out.contains("(result :ok 0 :error \"stack underflow\" :kind runtime)"),
        "envelope was: {}",
        out
    );
}

#[test]
fn test_sexp_eval_word_preserves_rest_of_line() {
    // The input-state guard: code after the word on the same line still runs.
    let mut vm = test_vm();
    let out = eval(&mut vm, "SEXP-EVAL\" (+ 2 3)\" 7 .");
    assert!(
        out.contains("(result :ok 1 :value (5)"),
        "envelope missing: {}",
        out
    );
    assert!(
        out.trim_end().ends_with('7'),
        "trailing '7 .' did not run: {}",
        out
    );
}

// --- Recruit handler (eval_sexp over the mesh recruit pattern) ---

#[test]
fn test_recruit_success_envelope() {
    let mut vm = test_vm();
    let mut m = under_ceiling;
    let reply = reply_of(vm.handle_recruit(7, 2, "(+ 2 3)", "parent", &mut m));
    let parsed = crate::sexp::parse(&reply).unwrap();
    let rr = crate::distgoal::read_recruit_result(&parsed).unwrap();
    // Routing fields round-trip (node_id_cache unset in test_vm -> "local").
    assert_eq!(rr.goal_id, 7);
    assert_eq!(rr.seq, 2);
    assert_eq!(rr.from, "local");
    assert_eq!(
        rr.result,
        crate::sexp::ResultView::Ok {
            value: vec![5],
            output: String::new()
        }
    );
}

#[test]
fn test_recruit_runtime_error_is_visible() {
    // A failing instr produces a VISIBLE :ok 0 envelope over the wire — unlike
    // the legacy sub-goal path, which silently trims output and drops failure.
    let mut vm = test_vm();
    let mut m = under_ceiling;
    let reply = reply_of(vm.handle_recruit(1, 0, "(drop)", "parent", &mut m));
    let rr =
        crate::distgoal::read_recruit_result(&crate::sexp::parse(&reply).unwrap()).unwrap();
    match rr.result {
        crate::sexp::ResultView::Err { kind, msg } => {
            assert_eq!(kind, "runtime");
            assert!(msg.contains("underflow"), "msg: {}", msg);
        }
        other => panic!("expected runtime error, got {:?}", other),
    }
}

#[test]
fn test_recruit_parse_error_is_visible() {
    let mut vm = test_vm();
    let mut m = under_ceiling;
    let reply = reply_of(vm.handle_recruit(1, 0, "(+ 2", "parent", &mut m)); // unterminated list
    let rr =
        crate::distgoal::read_recruit_result(&crate::sexp::parse(&reply).unwrap()).unwrap();
    match rr.result {
        crate::sexp::ResultView::Err { kind, .. } => assert_eq!(kind, "parse"),
        other => panic!("expected parse error, got {:?}", other),
    }
}

// --- Mechanical recruiter (emit + collect, full round-trip) ---

#[test]
fn test_recruiter_emits_wellformed() {
    let mut vm = test_vm();
    let msg = vm.send_recruit("peerABC", 4, 1, "(* 6 7)");
    // node_id_cache unset in test_vm -> :from "local".
    assert_eq!(msg, "(recruit :id 4 :seq 1 :from \"local\" :instr \"(* 6 7)\")");
    assert!(vm.recruit_ledger.is_pending(4, 1));
}

/// Drive a recruit message through the worker's handle_recruit and feed the
/// reply back through the recruiter's real collection path (process_chatter_msg).
fn loopback_recruit(recruiter: &mut VM, worker: &mut VM, goal_id: u64, seq: usize, instr: &str) {
    let recruit_msg = recruiter.send_recruit("worker", goal_id, seq, instr);
    let parsed = crate::sexp::parse(&recruit_msg).unwrap();
    let gid = parsed.get_key(":id").and_then(|s| s.as_number()).unwrap() as u64;
    let s = parsed.get_key(":seq").and_then(|s| s.as_number()).unwrap() as usize;
    let i = parsed.get_key(":instr").and_then(|s| s.as_str()).unwrap().to_string();
    let mut m = under_ceiling;
    let reply = reply_of(worker.handle_recruit(gid, s, &i, "recruiter", &mut m));
    recruiter.process_chatter_msg(&reply);
}

#[test]
fn test_recruit_round_trip_success() {
    let mut recruiter = test_vm();
    let mut worker = test_vm();
    loopback_recruit(&mut recruiter, &mut worker, 9, 0, "(+ 2 3)");
    let rr = recruiter
        .recruit_ledger
        .get(9, 0)
        .expect("reply should be collected");
    assert_eq!(
        rr.result,
        crate::sexp::ResultView::Ok {
            value: vec![5],
            output: String::new()
        }
    );
    assert!(!recruiter.recruit_ledger.is_pending(9, 0));
}

#[test]
fn test_recruit_round_trip_failure_visible_end_to_end() {
    // A faulting instruction's failure must be visible to the RECRUITER,
    // not just the worker — the collected envelope is :ok 0 with structured :error.
    let mut recruiter = test_vm();
    let mut worker = test_vm();
    loopback_recruit(&mut recruiter, &mut worker, 3, 1, "(drop)");
    let rr = recruiter
        .recruit_ledger
        .get(3, 1)
        .expect("reply should be collected");
    match &rr.result {
        crate::sexp::ResultView::Err { kind, msg } => {
            assert_eq!(kind, "runtime");
            assert!(msg.contains("underflow"), "msg: {}", msg);
        }
        other => panic!("expected runtime error, got {:?}", other),
    }
}

#[test]
fn test_recruit_word_fires_and_logs() {
    let mut vm = test_vm();
    let out = eval(&mut vm, "RECRUIT\" peerX (+ 2 3)\"");
    assert!(out.contains("(recruit :id"), "out: {}", out);
    assert!(out.contains(":instr \"(+ 2 3)\""), "out: {}", out);
    // The viewer word shows the now-pending request.
    let view = eval(&mut vm, "RECRUITS");
    assert!(view.contains("pending"), "view: {}", view);
}

// --- (parallel ...) split-and-recruit decision ---

// Injected resource readings (see commit/diff notes on the test seam): a real
// reading at any utilization via the crate-visible from_parts constructor — no
// memory exhaustion, no mock.
fn under_ceiling() -> crate::resources::HostResources {
    crate::resources::HostResources::from_parts(1000, 500, 0.0, 4) // util 0.5
}
fn over_ceiling() -> crate::resources::HostResources {
    crate::resources::HostResources::from_parts(1000, 50, 0.0, 4) // util 0.95
}

fn parse_parts(expr: &str) -> Vec<crate::sexp::Sexp> {
    crate::distgoal::parallel_parts(&crate::sexp::parse(expr).unwrap()).unwrap()
}

/// Unwrap a synchronous recruit reply, panicking if the unit deferred.
fn reply_of(o: crate::distgoal::RecruitOutcome) -> String {
    match o {
        crate::distgoal::RecruitOutcome::Reply(s) => s,
        other => panic!("expected synchronous Reply, got {:?}", other),
    }
}

/// Read a (parallel-result ...) into (overall_ok, ordered ResultViews).
fn read_parallel(sexp: &crate::sexp::Sexp) -> (i64, Vec<crate::sexp::ResultView>) {
    let ok = sexp.get_key(":ok").and_then(|s| s.as_number()).unwrap();
    let views = sexp
        .get_key(":results")
        .and_then(|s| s.as_list())
        .unwrap()
        .iter()
        .map(|e| crate::sexp::read_result(e).unwrap())
        .collect();
    (ok, views)
}

fn ok_values(views: &[crate::sexp::ResultView]) -> Vec<Vec<i64>> {
    views
        .iter()
        .map(|v| match v {
            crate::sexp::ResultView::Ok { value, .. } => value.clone(),
            other => panic!("expected ok, got {:?}", other),
        })
        .collect()
}

#[test]
fn test_parallel_all_local_with_headroom() {
    let mut vm = test_vm();
    let parts = parse_parts("(parallel (+ 1 1) (+ 2 2) (+ 3 3))");
    let mut measure = under_ceiling;
    let goal_id = vm.run_parallel(&parts, &mut measure, 0);
    let (ok, views) = read_parallel(&vm.parallel_result(goal_id).unwrap());
    assert_eq!(ok, 1);
    assert_eq!(ok_values(&views), vec![vec![2], vec![4], vec![6]]);
}

#[test]
fn test_parallel_forced_recruit_round_trip() {
    // Over ceiling for every part -> all recruited, all slots pending. Loopback
    // each through a worker's handle_recruit and the recruiter's collection arm.
    let mut recruiter = test_vm();
    let mut worker = test_vm();
    let parts = parse_parts("(parallel (+ 1 1) (+ 2 2) (+ 3 3))");
    let mut measure = over_ceiling;
    let goal_id = recruiter.run_parallel(&parts, &mut measure, 0);
    // Before any reply, the job is incomplete -> overall :ok 0.
    assert_eq!(
        read_parallel(&recruiter.parallel_result(goal_id).unwrap()).0,
        0
    );
    for (seq, part) in parts.iter().enumerate() {
        let mut wm = under_ceiling;
        let reply = reply_of(worker.handle_recruit(goal_id, seq, &part.to_string(), "recruiter", &mut wm));
        recruiter.process_chatter_msg(&reply);
    }
    let (ok, views) = read_parallel(&recruiter.parallel_result(goal_id).unwrap());
    assert_eq!(ok, 1);
    assert_eq!(ok_values(&views), vec![vec![2], vec![4], vec![6]]);
}

#[test]
fn test_parallel_mixed_local_and_recruited_preserves_order() {
    let mut recruiter = test_vm();
    let mut worker = test_vm();
    let parts = parse_parts("(parallel (+ 1 1) (+ 2 2) (+ 3 3))");
    // Measure per part: part 0 local, part 1 recruited, part 2 local.
    let mut call = 0;
    let mut measure = || {
        call += 1;
        if call == 2 {
            over_ceiling()
        } else {
            under_ceiling()
        }
    };
    let goal_id = recruiter.run_parallel(&parts, &mut measure, 0);
    // Only the middle part is outstanding; loopback just that one.
    let mut wm = under_ceiling;
    let reply = reply_of(worker.handle_recruit(goal_id, 1, &parts[1].to_string(), "recruiter", &mut wm));
    recruiter.process_chatter_msg(&reply);
    let (ok, views) = read_parallel(&recruiter.parallel_result(goal_id).unwrap());
    assert_eq!(ok, 1);
    // Order preserved across the local/recruited boundary.
    assert_eq!(ok_values(&views), vec![vec![2], vec![4], vec![6]]);
}

#[test]
fn test_parallel_failure_is_visible_in_collected_results() {
    let mut vm = test_vm();
    // Middle part underflows.
    let parts = parse_parts("(parallel (+ 1 1) (drop) (+ 3 3))");
    let mut measure = under_ceiling;
    let goal_id = vm.run_parallel(&parts, &mut measure, 0);
    let (ok, views) = read_parallel(&vm.parallel_result(goal_id).unwrap());
    assert_eq!(ok, 0, "overall ok must be 0 when a sub-part faults");
    match &views[1] {
        crate::sexp::ResultView::Err { kind, msg } => {
            assert_eq!(kind, "runtime");
            assert!(msg.contains("underflow"), "msg: {}", msg);
        }
        other => panic!("expected runtime error in slot 1, got {:?}", other),
    }
    // Flanking parts still succeeded.
    assert!(matches!(views[0], crate::sexp::ResultView::Ok { .. }));
    assert!(matches!(views[2], crate::sexp::ResultView::Ok { .. }));
}

#[test]
fn test_parallel_single_element_degenerate() {
    let mut vm = test_vm();
    let parts = parse_parts("(parallel (+ 2 3))");
    assert_eq!(parts.len(), 1);
    let mut measure = under_ceiling;
    let goal_id = vm.run_parallel(&parts, &mut measure, 0);
    let (ok, views) = read_parallel(&vm.parallel_result(goal_id).unwrap());
    assert_eq!(ok, 1);
    assert_eq!(ok_values(&views), vec![vec![5]]);
}

#[test]
fn test_parallel_word_fires() {
    // Uses live HostResources::measure, so don't assert local-vs-recruit (that
    // depends on machine state); just that the word produces a well-formed
    // parallel-result with one sub-result envelope per part.
    let mut vm = test_vm();
    let out = eval(&mut vm, "PARALLEL\" (parallel (+ 1 1) (+ 2 2))\"");
    assert!(out.contains("(parallel-result :ok"), "out: {}", out);
    assert_eq!(
        out.matches("(result :ok").count(),
        2,
        "expected 2 sub-result envelopes, out: {}",
        out
    );
}

// --- (alloc-mb N) load generator ---

#[test]
fn test_alloc_mb_refuses_when_disabled() {
    // Default gate is off: ALLOC-MB must NOT allocate, and pushes 0.
    let mut vm = test_vm();
    assert!(!vm.alloc_enabled, "gate must default off");
    let env = crate::sexp::eval_sexp(&mut vm, "(alloc-mb 4)");
    match crate::sexp::read_result(&env).unwrap() {
        crate::sexp::ResultView::Ok { value, output } => {
            assert_eq!(value, vec![0], "refused -> push 0");
            assert!(output.contains("disabled"), "should report disabled: {}", output);
        }
        other => panic!("expected ok envelope, got {:?}", other),
    }
    assert!(vm.mem_ballast.is_empty(), "no allocation when disabled");
}

#[test]
fn test_alloc_enable_word_then_allocates() {
    let mut vm = test_vm();
    // The enable word toggles the gate (mirrors SHELL-ENABLE).
    let out = vm.eval("ALLOC-ENABLE");
    assert!(out.contains("ENABLED"), "enable status: {}", out);
    assert!(vm.alloc_enabled);
    let env = crate::sexp::eval_sexp(&mut vm, "(alloc-mb 2)");
    assert_eq!(
        crate::sexp::read_result(&env),
        Some(crate::sexp::ResultView::Ok {
            value: vec![2],
            output: String::new()
        })
    );
    assert_eq!(vm.mem_ballast.len(), 1);
    vm.eval("RECLAIM-MB");
}

#[test]
fn test_alloc_mb_accumulates() {
    // Each ALLOC-MB APPENDS a buffer (accumulate), so parts stack pressure —
    // they don't replace a single buffer.
    let mut vm = test_vm();
    vm.alloc_enabled = true;
    crate::sexp::eval_sexp(&mut vm, "(alloc-mb 1)");
    crate::sexp::eval_sexp(&mut vm, "(alloc-mb 1)");
    assert_eq!(vm.mem_ballast.len(), 2, "two calls must stack, not replace");
    let total: usize = vm.mem_ballast.iter().map(|b| b.len()).sum();
    assert_eq!(total, 2 * 1024 * 1024, "total retained ~= sum of calls");
    // RECLAIM-MB frees ALL accumulated buffers.
    vm.eval("RECLAIM-MB");
    assert!(vm.mem_ballast.is_empty());
}

#[test]
fn test_reclaim_works_regardless_of_gate() {
    // Freeing must never be blocked, even with the gate off.
    let mut vm = test_vm();
    vm.alloc_enabled = true;
    crate::sexp::eval_sexp(&mut vm, "(alloc-mb 1)");
    assert_eq!(vm.mem_ballast.len(), 1);
    // Disable the gate, then RECLAIM-MB must still free the retained memory.
    vm.alloc_enabled = false;
    let out = vm.eval("RECLAIM-MB ."); // prints the freed-chunk count
    assert_eq!(out.trim(), "1", "freed 1 chunk: {}", out);
    assert!(vm.mem_ballast.is_empty(), "memory recovered despite gate off");
}

#[test]
fn test_alloc_mb_through_eval_sexp() {
    // (alloc-mb N) round-trips through the seam: it runs, retains memory, and
    // returns the MiB allocated as its (trivial) result value.
    let mut vm = test_vm();
    vm.alloc_enabled = true; // gated off by default
    let env = crate::sexp::eval_sexp(&mut vm, "(alloc-mb 2)");
    assert_eq!(
        crate::sexp::read_result(&env),
        Some(crate::sexp::ResultView::Ok {
            value: vec![2],
            output: String::new()
        })
    );
    assert!(!vm.mem_ballast.is_empty(), "memory should be retained");
    // RECLAIM-MB releases it.
    vm.eval("RECLAIM-MB");
    assert!(vm.mem_ballast.is_empty(), "memory should be freed");
}

#[test]
fn test_parallel_of_alloc_mb_parts_runs() {
    // The point of the parts is to consume resources, not produce a value; they
    // still run, return, and assemble in order. Small sizes keep the test cheap.
    let mut vm = test_vm();
    vm.alloc_enabled = true; // gated off by default
    let parts = parse_parts("(parallel (alloc-mb 1) (alloc-mb 1) (alloc-mb 1))");
    let mut measure = under_ceiling; // force all-local so the parts actually run
    let goal_id = vm.run_parallel(&parts, &mut measure, 0);
    let (ok, views) = read_parallel(&vm.parallel_result(goal_id).unwrap());
    assert_eq!(ok, 1);
    assert_eq!(ok_values(&views), vec![vec![1], vec![1], vec![1]]);
    // Three 1 MiB chunks retained; reclaim so the test process doesn't keep them.
    assert_eq!(vm.mem_ballast.len(), 3);
    vm.eval("RECLAIM-MB");
    assert!(vm.mem_ballast.is_empty());
}

// --- Committed-work accounting in run_parallel admission (per-call) ---

// A measure() PINNED at 50% utilization over a 10 MiB budget: it never changes,
// so ONLY the committed tally can shift a run-vs-recruit decision.
fn pinned_half() -> crate::resources::HostResources {
    crate::resources::HostResources::from_parts(10_240, 5_120, 0.0, 4) // util 0.5
}
const PIN_BUDGET_KB: u64 = 10_240; // 10 MiB; each (alloc-mb 1) = 1/10 of budget

#[test]
fn test_committed_tally_recruits_later_parts_under_pinned_measure() {
    let mut vm = test_vm();
    vm.alloc_enabled = true;
    let mut measure = pinned_half;
    // 0.50 observed + committed: parts 0,1,2 fit (0.50, 0.60, 0.70 < 0.80);
    // part 3 would be 0.80 -> recruit; part 4 -> recruit. measure() never moves,
    // so it is the committed tally — not a change in measure() — that recruits.
    let parts = parse_parts("(parallel (alloc-mb 1) (alloc-mb 1) (alloc-mb 1) (alloc-mb 1) (alloc-mb 1))");
    let goal_id = vm.run_parallel(&parts, &mut measure, PIN_BUDGET_KB);
    assert_eq!(
        vm.mem_ballast.len(),
        3,
        "committed tally must stop local runs at 3 (raw measure alone never would)"
    );
    let (ok, _) = read_parallel(&vm.parallel_result(goal_id).unwrap());
    assert_eq!(ok, 0, "the 2 recruited parts stay pending -> overall not ok");
    vm.eval("RECLAIM-MB");
}

#[test]
fn test_committed_tally_resets_between_calls() {
    let mut vm = test_vm();
    vm.alloc_enabled = true;
    let mut measure = pinned_half;
    // First parallel fills the tally (3 local, 2 recruited).
    let p1 = parse_parts("(parallel (alloc-mb 1) (alloc-mb 1) (alloc-mb 1) (alloc-mb 1) (alloc-mb 1))");
    vm.run_parallel(&p1, &mut measure, PIN_BUDGET_KB);
    vm.eval("RECLAIM-MB");
    // Second parallel starts from a CLEAN tally: a single part runs locally,
    // not penalized by the first call's committed work.
    let p2 = parse_parts("(parallel (alloc-mb 1))");
    let g2 = vm.run_parallel(&p2, &mut measure, PIN_BUDGET_KB);
    let (ok, views) = read_parallel(&vm.parallel_result(g2).unwrap());
    assert_eq!(ok, 1, "second call's single part runs locally (tally reset)");
    assert_eq!(ok_values(&views), vec![vec![1]]);
    vm.eval("RECLAIM-MB");
}

#[test]
fn test_committed_single_part_runs_local_no_false_recruit() {
    let mut vm = test_vm();
    vm.alloc_enabled = true;
    let mut measure = pinned_half;
    // 0.50 + 0 committed < 0.80 -> the single part runs locally.
    let parts = parse_parts("(parallel (alloc-mb 1))");
    let g = vm.run_parallel(&parts, &mut measure, PIN_BUDGET_KB);
    let (ok, _) = read_parallel(&vm.parallel_result(g).unwrap());
    assert_eq!(ok, 1);
    assert_eq!(vm.mem_ballast.len(), 1);
    vm.eval("RECLAIM-MB");
}

#[test]
fn test_committed_unknown_cost_parts_contribute_zero() {
    let mut vm = test_vm();
    // Arithmetic parts have no cost model -> contribute 0, so even five of them
    // all run locally under the pinned under-ceiling measure (no false recruit
    // from a fabricated cost).
    let mut measure = pinned_half;
    let parts = parse_parts("(parallel (+ 1 1) (+ 2 2) (+ 3 3) (+ 4 4) (+ 5 5))");
    let g = vm.run_parallel(&parts, &mut measure, PIN_BUDGET_KB);
    let (ok, views) = read_parallel(&vm.parallel_result(g).unwrap());
    assert_eq!(ok, 1, "unknown-cost parts contribute 0 -> all run locally");
    assert_eq!(views.len(), 5);
}

// --- Recursive recruit fan-out (the decision recurses, ceiling-bounded) ---

#[test]
fn test_recursive_two_level_fanout() {
    // Parts are THEMSELVES (parallel ...) forms.
    let mut recruiter = test_vm();
    let mut worker = test_vm();
    let parts = parse_parts("(parallel (parallel (+ 1 1) (+ 2 2)) (parallel (+ 3 3)))");
    // Level 1: no headroom -> both parts recruited (no mesh here -> pending; we
    // loopback as the recruited worker).
    let mut l1 = over_ceiling;
    let goal_id = recruiter.run_parallel(&parts, &mut l1, 0);
    // The recruited worker HAS headroom, so when it receives a (parallel ...)
    // part it runs its OWN run_parallel and handles the sub-parts locally.
    for (seq, part) in parts.iter().enumerate() {
        let mut wm = under_ceiling;
        let reply = reply_of(worker.handle_recruit(goal_id, seq, &part.to_string(), "recruiter", &mut wm));
        recruiter.process_chatter_msg(&reply);
    }
    let result = recruiter.parallel_result(goal_id).unwrap();
    assert_eq!(result.get_key(":ok").and_then(|s| s.as_number()), Some(1));
    // Each parent slot holds a NESTED (parallel-result ...), ordered, not flattened.
    let nested = result.get_key(":results").and_then(|s| s.as_list()).unwrap();
    assert_eq!(nested.len(), 2);
    assert_eq!(crate::sexp::msg_type(&nested[0]), Some("parallel-result"));
    assert_eq!(crate::sexp::msg_type(&nested[1]), Some("parallel-result"));
    // First nested has two sub-results (2, 4); second has one (6).
    let first = nested[0].get_key(":results").and_then(|s| s.as_list()).unwrap();
    let second = nested[1].get_key(":results").and_then(|s| s.as_list()).unwrap();
    assert_eq!(ok_values(&first.iter().map(|e| crate::sexp::read_result(e).unwrap()).collect::<Vec<_>>()), vec![vec![2], vec![4]]);
    assert_eq!(ok_values(&second.iter().map(|e| crate::sexp::read_result(e).unwrap()).collect::<Vec<_>>()), vec![vec![6]]);
}

#[test]
fn test_recursive_fanout_halts_when_no_peer_has_headroom() {
    // THE BRAKE. No mesh -> choose_recruit_peer returns None (the same boundary
    // a saturated mesh hits, where transport::choose_destination returns None —
    // see choose_destination_none_when_no_peer_sufficient). With no headroom and
    // no choosable peer, run_parallel must terminate and emit ZERO recruits;
    // parts are declined (slots stay pending). This is emergent termination:
    // the ceiling check, present at every level, stops the tree growing.
    let mut vm = test_vm();
    let parts = parse_parts("(parallel (+ 1 1) (+ 2 2) (+ 3 3))");
    let mut measure = over_ceiling; // no headroom
    let goal_id = vm.run_parallel(&parts, &mut measure, 0);
    // Terminated (we're here, not looping). No recruit emitted.
    assert_eq!(
        vm.recruit_ledger.len(),
        0,
        "a peerless/saturated mesh must emit no recruits"
    );
    // Parts declined: all slots pending -> overall :ok 0.
    let (ok, _) = read_parallel(&vm.parallel_result(goal_id).unwrap());
    assert_eq!(ok, 0);
}

// --- Fan-in: report-once-when-complete result propagation up the tree ---

// Helpers to read a recruit-result wrapper's nested envelope and routing.
fn nested_of(msg: &str) -> crate::sexp::Sexp {
    crate::sexp::parse(msg)
        .unwrap()
        .get_key(":result")
        .cloned()
        .unwrap()
}
fn leaf_values(parallel_result: &crate::sexp::Sexp) -> Vec<Vec<i64>> {
    let leaves = parallel_result
        .get_key(":results")
        .and_then(|s| s.as_list())
        .unwrap();
    ok_values(
        &leaves
            .iter()
            .map(|e| crate::sexp::read_result(e).unwrap())
            .collect::<Vec<_>>(),
    )
}

#[test]
fn test_deep_completion_propagates_up() {
    // Three roles: root recruits a (parallel ...) part to middle; middle has
    // headroom for only ONE of its sub-parts, recruits the other to grandchild.
    // The root's result must come back COMPLETE only after the grandchild's
    // result propagates up THROUGH middle. This is the fan-in that was broken.
    let mut root = test_vm();
    let mut middle = test_vm();
    let mut grand = test_vm();

    let parts = parse_parts("(parallel (parallel (+ 1 1) (+ 2 2)))");
    let mut root_m = over_ceiling; // root: no headroom -> recruit the part
    let root_g = root.run_parallel(&parts, &mut root_m, 0);
    assert_eq!(read_parallel(&root.parallel_result(root_g).unwrap()).0, 0); // pending

    // Middle: headroom for sub-part 0 only -> runs it, recruits sub-part 1.
    let mut mid_call = 0;
    let mut mid_m = || {
        mid_call += 1;
        if mid_call == 1 {
            under_ceiling()
        } else {
            over_ceiling()
        }
    };
    let mid_child = match middle.handle_recruit(root_g, 0, &parts[0].to_string(), "root", &mut mid_m)
    {
        crate::distgoal::RecruitOutcome::Deferred { child_goal_id } => child_goal_id,
        other => panic!("middle recruited overflow, should defer; got {:?}", other),
    };
    // Back-reference stored; middle has NOT reported up yet.
    assert!(middle.report_targets.contains_key(&mid_child));

    // Grandchild computes middle's overflow sub-part (seq 1 -> (+ 2 2)).
    let mut g_m = under_ceiling;
    let g_reply = reply_of(grand.handle_recruit(mid_child, 1, "(+ 2 2)", "middle", &mut g_m));

    // Middle collects it -> middle now whole -> self-reports UP to root.
    let (target, mid_report) = middle
        .collect_recruit_result(&crate::sexp::parse(&g_reply).unwrap())
        .expect("middle should self-report on completion");
    assert_eq!(target, "root"); // the back-reference is used to target the report

    // Root collects middle's COMPLETE report -> root is whole (it is the root).
    assert!(
        root.collect_recruit_result(&crate::sexp::parse(&mid_report).unwrap())
            .is_none(),
        "root surfaces locally, does not report further"
    );
    let result = root.parallel_result(root_g).unwrap();
    assert_eq!(
        result.get_key(":ok").and_then(|s| s.as_number()),
        Some(1),
        "root result must be COMPLETE (no pending holes)"
    );
    let nested = result.get_key(":results").and_then(|s| s.as_list()).unwrap();
    assert_eq!(crate::sexp::msg_type(&nested[0]), Some("parallel-result"));
    assert_eq!(leaf_values(&nested[0]), vec![vec![2], vec![4]]);
}

#[test]
fn test_last_slot_fill_triggers_self_report() {
    // Middle recruits BOTH sub-parts (no headroom). Filling the first slot must
    // NOT self-report; filling the last slot MUST, targeting the back-reference.
    let mut middle = test_vm();
    let mut grand = test_vm();
    let parts = parse_parts("(parallel (parallel (+ 1 1) (+ 2 2)))");
    let mut mid_m = over_ceiling;
    let mid_child =
        match middle.handle_recruit(99, 5, &parts[0].to_string(), "the-recruiter", &mut mid_m) {
            crate::distgoal::RecruitOutcome::Deferred { child_goal_id } => child_goal_id,
            other => panic!("expected deferred, got {:?}", other),
        };
    let mut g_m = under_ceiling;
    let r0 = reply_of(grand.handle_recruit(mid_child, 0, "(+ 1 1)", "middle", &mut g_m));
    let r1 = reply_of(grand.handle_recruit(mid_child, 1, "(+ 2 2)", "middle", &mut g_m));

    // First (non-last) slot: no self-report.
    assert!(middle
        .collect_recruit_result(&crate::sexp::parse(&r0).unwrap())
        .is_none());
    // Last slot: self-report, targeting the stored back-reference + routing.
    let (target, msg) = middle
        .collect_recruit_result(&crate::sexp::parse(&r1).unwrap())
        .unwrap();
    assert_eq!(target, "the-recruiter");
    let report = crate::sexp::parse(&msg).unwrap();
    assert_eq!(report.get_key(":id").and_then(|s| s.as_number()), Some(99));
    assert_eq!(report.get_key(":seq").and_then(|s| s.as_number()), Some(5));
}

#[test]
fn test_synchronous_completion_still_replies() {
    // A recruited (parallel ...) that completes fully locally still replies
    // synchronously — no back-reference, no async path (prior behavior intact).
    let mut worker = test_vm();
    let mut m = under_ceiling; // headroom for everything
    let outcome = worker.handle_recruit(1, 0, "(parallel (+ 1 1) (+ 2 2))", "recruiter", &mut m);
    match outcome {
        crate::distgoal::RecruitOutcome::Reply(reply) => {
            let nested = nested_of(&reply);
            assert_eq!(crate::sexp::msg_type(&nested), Some("parallel-result"));
            assert_eq!(nested.get_key(":ok").and_then(|s| s.as_number()), Some(1));
        }
        other => panic!("expected synchronous Reply, got {:?}", other),
    }
    assert!(
        worker.report_targets.is_empty(),
        "no back-reference for a synchronously-completed job"
    );
}

#[test]
fn test_deep_failure_propagates_up() {
    // A fault in a grandchild part surfaces :ok 0 :kind runtime in the nested
    // envelope and forces :ok 0 up through each parent's COMPLETE result.
    let mut root = test_vm();
    let mut middle = test_vm();
    let mut grand = test_vm();
    let parts = parse_parts("(parallel (parallel (+ 1 1) (drop)))");
    let mut root_m = over_ceiling;
    let root_g = root.run_parallel(&parts, &mut root_m, 0);
    let mut mid_call = 0;
    let mut mid_m = || {
        mid_call += 1;
        if mid_call == 1 {
            under_ceiling()
        } else {
            over_ceiling()
        }
    };
    let mid_child = match middle.handle_recruit(root_g, 0, &parts[0].to_string(), "root", &mut mid_m)
    {
        crate::distgoal::RecruitOutcome::Deferred { child_goal_id } => child_goal_id,
        other => panic!("expected deferred, got {:?}", other),
    };
    // Grandchild runs the faulting overflow part (drop -> underflow).
    let mut g_m = under_ceiling;
    let g_reply = reply_of(grand.handle_recruit(mid_child, 1, "(drop)", "middle", &mut g_m));
    let (_, mid_report) = middle
        .collect_recruit_result(&crate::sexp::parse(&g_reply).unwrap())
        .unwrap();
    root.collect_recruit_result(&crate::sexp::parse(&mid_report).unwrap());

    let result = root.parallel_result(root_g).unwrap();
    assert_eq!(result.get_key(":ok").and_then(|s| s.as_number()), Some(0));
    let nested = result.get_key(":results").and_then(|s| s.as_list()).unwrap();
    assert_eq!(nested[0].get_key(":ok").and_then(|s| s.as_number()), Some(0));
    let leaves = nested[0].get_key(":results").and_then(|s| s.as_list()).unwrap();
    match crate::sexp::read_result(&leaves[1]).unwrap() {
        crate::sexp::ResultView::Err { kind, .. } => assert_eq!(kind, "runtime"),
        other => panic!("expected runtime error leaf, got {:?}", other),
    }
}

#[test]
fn test_ordering_preserved_across_out_of_order_fill() {
    // Three recruited sub-parts; their results arrive OUT OF ORDER (2,0,1). The
    // reported parallel-result must still list results in SLOT order.
    let mut middle = test_vm();
    let mut grand = test_vm();
    let parts = parse_parts("(parallel (parallel (+ 1 1) (+ 2 2) (+ 3 3)))");
    let mut mid_m = over_ceiling;
    let mid_child = match middle.handle_recruit(7, 0, &parts[0].to_string(), "root", &mut mid_m) {
        crate::distgoal::RecruitOutcome::Deferred { child_goal_id } => child_goal_id,
        other => panic!("expected deferred, got {:?}", other),
    };
    let mut g_m = under_ceiling;
    let r0 = reply_of(grand.handle_recruit(mid_child, 0, "(+ 1 1)", "m", &mut g_m));
    let r1 = reply_of(grand.handle_recruit(mid_child, 1, "(+ 2 2)", "m", &mut g_m));
    let r2 = reply_of(grand.handle_recruit(mid_child, 2, "(+ 3 3)", "m", &mut g_m));
    // Fill out of order: 2, then 0, then 1 (the last fill completes the job).
    assert!(middle
        .collect_recruit_result(&crate::sexp::parse(&r2).unwrap())
        .is_none());
    assert!(middle
        .collect_recruit_result(&crate::sexp::parse(&r0).unwrap())
        .is_none());
    let (_, msg) = middle
        .collect_recruit_result(&crate::sexp::parse(&r1).unwrap())
        .unwrap();
    // Results ordered by slot despite out-of-order arrival.
    assert_eq!(leaf_values(&nested_of(&msg)), vec![vec![2], vec![4], vec![6]]);
}

// --- Supervision: re-recruit on peer death (gossip-death, let-it-crash) ---

fn addr(s: &str) -> std::net::SocketAddr {
    s.parse().unwrap()
}

#[test]
fn test_resupervise_re_recruits_on_peer_death() {
    let mut parent = test_vm();
    // Parent recruited slot (5,0)="(+ 2 2)" to peer "deadpeer"; tracked as a
    // 1-part parallel job (what run_parallel would have set up).
    parent
        .parallel_jobs
        .insert(5, crate::distgoal::ParallelJob::new(5, 1));
    parent.send_recruit("deadpeer", 5, 0, "(+ 2 2)");
    assert_eq!(parent.recruit_ledger.holder(5, 0), Some("deadpeer"));

    // "deadpeer" leaves the gossip view; an abundant peer "Q" is live.
    parent.supervise_recruits(&[("Q".to_string(), 90, addr("127.0.0.1:9001"))]);

    // Re-recruited to Q: holder reassigned, instruction retained, still pending.
    assert_eq!(parent.recruit_ledger.holder(5, 0), Some("Q"));
    assert_eq!(parent.recruit_ledger.pending_instr(5, 0), Some("(+ 2 2)"));
    assert!(parent.recruit_ledger.is_pending(5, 0));

    // Q reports -> slot fills -> parent completes normally.
    let mut worker = test_vm();
    let mut m = under_ceiling;
    let q_reply = reply_of(worker.handle_recruit(5, 0, "(+ 2 2)", "parent", &mut m));
    assert!(parent
        .collect_recruit_result(&crate::sexp::parse(&q_reply).unwrap())
        .is_none());
    let (ok, views) = read_parallel(&parent.parallel_result(5).unwrap());
    assert_eq!(ok, 1);
    assert_eq!(ok_values(&views), vec![vec![4]]);
    // Instruction released after the slot settles.
    assert!(!parent.recruit_ledger.is_pending(5, 0));
    assert_eq!(parent.recruit_ledger.pending_instr(5, 0), None);
}

#[test]
fn test_ledger_retains_then_releases_instruction() {
    let mut parent = test_vm();
    parent.send_recruit("P", 3, 1, "(* 6 7)");
    // Retained while pending.
    assert_eq!(parent.recruit_ledger.pending_instr(3, 1), Some("(* 6 7)"));
    assert!(parent.recruit_ledger.is_pending(3, 1));
    // Fill it.
    let env = crate::sexp::msg_result(crate::sexp::EvalOutcome::Ok {
        stack: &[42],
        output: "",
    });
    let reply = crate::distgoal::sexp_recruit_result(3, 1, "P", &env);
    parent.collect_recruit_result(&crate::sexp::parse(&reply).unwrap());
    // Released after fill; result settled.
    assert_eq!(parent.recruit_ledger.pending_instr(3, 1), None);
    assert!(!parent.recruit_ledger.is_pending(3, 1));
}

#[test]
fn test_resupervise_fail_closed_when_no_headroom_peer() {
    let mut parent = test_vm();
    parent.send_recruit("deadpeer", 7, 0, "(+ 1 1)");
    // deadpeer dies; the only live peer "Z" has NO headroom (5% < sufficiency).
    parent.supervise_recruits(&[("Z".to_string(), 5, addr("127.0.0.1:9002"))]);
    // Fail-closed: no re-recruit, slot stays held by deadpeer and pending.
    assert!(parent.recruit_ledger.is_pending(7, 0));
    assert_eq!(parent.recruit_ledger.holder(7, 0), Some("deadpeer"));
    assert_eq!(parent.recruit_ledger.pending_instr(7, 0), Some("(+ 1 1)"));
}

#[test]
fn test_recursive_supervision_re_recruits_whole_subtree() {
    let mut root = test_vm();
    // Root recruited a (parallel ...) SUBTREE to "middle".
    root.parallel_jobs
        .insert(2, crate::distgoal::ParallelJob::new(2, 1));
    root.send_recruit("middle", 2, 0, "(parallel (+ 1 1) (+ 2 2))");

    // "middle" dies; a fresh peer "M2" with headroom exists.
    root.supervise_recruits(&[("M2".to_string(), 90, addr("127.0.0.1:9003"))]);
    // The WHOLE subtree instruction was re-recruited to M2 (not just a leaf).
    assert_eq!(root.recruit_ledger.holder(2, 0), Some("M2"));
    assert_eq!(
        root.recruit_ledger.pending_instr(2, 0),
        Some("(parallel (+ 1 1) (+ 2 2))")
    );

    // M2 re-runs the subtree (has headroom -> completes locally) and reports.
    let mut m2 = test_vm();
    let mut m = under_ceiling;
    let m2_reply = reply_of(m2.handle_recruit(2, 0, "(parallel (+ 1 1) (+ 2 2))", "root", &mut m));
    root.collect_recruit_result(&crate::sexp::parse(&m2_reply).unwrap());
    let result = root.parallel_result(2).unwrap();
    assert_eq!(result.get_key(":ok").and_then(|s| s.as_number()), Some(1));
    let nested = result.get_key(":results").and_then(|s| s.as_list()).unwrap();
    assert_eq!(crate::sexp::msg_type(&nested[0]), Some("parallel-result"));
    assert_eq!(leaf_values(&nested[0]), vec![vec![2], vec![4]]);
}

#[test]
fn test_supervision_leaves_healthy_slots_alone() {
    let mut parent = test_vm();
    parent
        .parallel_jobs
        .insert(8, crate::distgoal::ParallelJob::new(8, 1));
    parent.send_recruit("P", 8, 0, "(+ 3 4)");
    // P is ALIVE in the view -> supervision must not touch the slot.
    parent.supervise_recruits(&[("P".to_string(), 90, addr("127.0.0.1:9004"))]);
    assert_eq!(parent.recruit_ledger.holder(8, 0), Some("P")); // unchanged
    assert_eq!(parent.recruit_ledger.pending_instr(8, 0), Some("(+ 3 4)"));

    // P reports normally -> completes exactly as before.
    let mut worker = test_vm();
    let mut m = under_ceiling;
    let reply = reply_of(worker.handle_recruit(8, 0, "(+ 3 4)", "parent", &mut m));
    parent.collect_recruit_result(&crate::sexp::parse(&reply).unwrap());
    let (ok, views) = read_parallel(&parent.parallel_result(8).unwrap());
    assert_eq!(ok, 1);
    assert_eq!(ok_values(&views), vec![vec![7]]);
}

#[test]
fn test_recursive_failure_propagates_through_two_levels() {
    // A fault in a deeply-nested sub-part surfaces as :ok 0 :kind runtime in the
    // nested envelope and forces :ok 0 up through each parent parallel-result.
    let mut recruiter = test_vm();
    let mut worker = test_vm();
    let parts = parse_parts("(parallel (parallel (+ 1 1) (drop)))");
    let mut l1 = over_ceiling;
    let goal_id = recruiter.run_parallel(&parts, &mut l1, 0);
    let mut wm = under_ceiling;
    let reply = reply_of(worker.handle_recruit(goal_id, 0, &parts[0].to_string(), "recruiter", &mut wm));
    recruiter.process_chatter_msg(&reply);

    let result = recruiter.parallel_result(goal_id).unwrap();
    // Top level :ok 0.
    assert_eq!(result.get_key(":ok").and_then(|s| s.as_number()), Some(0));
    let nested = result.get_key(":results").and_then(|s| s.as_list()).unwrap();
    // Nested parallel-result :ok 0.
    assert_eq!(nested[0].get_key(":ok").and_then(|s| s.as_number()), Some(0));
    // The faulting leaf is a runtime error envelope.
    let leaves = nested[0].get_key(":results").and_then(|s| s.as_list()).unwrap();
    match crate::sexp::read_result(&leaves[1]).unwrap() {
        crate::sexp::ResultView::Err { kind, .. } => assert_eq!(kind, "runtime"),
        other => panic!("expected runtime error leaf, got {:?}", other),
    }
}

#[test]
fn test_vm_stack_top() {
    let mut vm = test_vm();
    assert_eq!(vm.stack_top(), None);
    eval(&mut vm, "42");
    assert_eq!(vm.stack_top(), Some(42));
}

#[test]
fn test_vm_eval_multiple_calls() {
    let mut vm = test_vm();
    eval(&mut vm, ": DOUBLE 2 * ;");
    assert_eq!(eval_top(&mut vm, "5 DOUBLE"), 10);
    vm.stack.clear();
    assert_eq!(eval_top(&mut vm, "3 DOUBLE"), 6);
}

#[test]
fn test_vm_isolation() {
    let mut vm1 = test_vm();
    let mut vm2 = test_vm();
    eval(&mut vm1, ": FOO 42 ;");
    eval(&mut vm1, "FOO");
    assert_eq!(vm1.stack_top(), Some(42));
    // vm2 should NOT have FOO
    eval(&mut vm2, "FOO"); // prints "FOO?" but doesn't crash
    assert_eq!(vm2.stack_top(), None); // stack empty — FOO not found
}

#[test]
fn test_vm_output_buffer() {
    let mut vm = test_vm();
    vm.output_buffer = Some(String::new());
    vm.interpret_line(".\" hello\"");
    let out = vm.output_buffer.take().unwrap();
    assert_eq!(out, "hello");
}

#[test]
fn test_vm_find_word() {
    let vm = test_vm();
    assert!(vm.find_word("DUP").is_some());
    assert!(vm.find_word("NONEXISTENT_WORD_XYZ").is_none());
}

#[test]
fn test_vm_prelude_loaded() {
    let vm = test_vm();
    for name in &["NIP", "TUCK", "ABS", "MIN", "MAX", "TRUE", "FALSE"] {
        assert!(
            vm.find_word(name).is_some(),
            "prelude word '{}' missing",
            name
        );
    }
}

#[test]
fn test_vm_eval_error_output() {
    let mut vm = test_vm();
    // Evaluating an unknown word should not crash, stack stays intact
    eval(&mut vm, "42");
    eval(&mut vm, "NONEXISTENT_WORD");
    assert_eq!(vm.stack_top(), Some(42)); // stack preserved
}

#[test]
fn test_vm_complex_program() {
    let mut vm = test_vm();
    eval(
        &mut vm,
        ": FIB DUP 2 < IF DROP 1 ELSE DUP 1 - RECURSE SWAP 2 - RECURSE + THEN ;",
    );
    assert_eq!(eval_top(&mut vm, "10 FIB"), 89);
}

#[test]
fn test_here_and_comma() {
    let mut vm = test_vm();
    let h1 = eval_top(&mut vm, "HERE");
    eval(&mut vm, "42 ,");
    eval(&mut vm, "99 ,");
    let h2 = eval_top(&mut vm, "HERE");
    assert_eq!(h2, h1 + 2);
    // Fetch stored values
    vm.stack.clear();
    vm.stack.push(h1);
    eval(&mut vm, "@");
    assert_eq!(vm.stack_top(), Some(42));
    vm.stack.clear();
    vm.stack.push(h1 + 1);
    eval(&mut vm, "@");
    assert_eq!(vm.stack_top(), Some(99));
}

#[test]
fn test_allot() {
    let mut vm = test_vm();
    let h1 = eval_top(&mut vm, "HERE");
    eval(&mut vm, "10 ALLOT");
    let h2 = eval_top(&mut vm, "HERE");
    assert_eq!(h2, h1 + 10);
}

#[test]
fn test_create_with_comma() {
    let mut vm = test_vm();
    eval(&mut vm, "CREATE MYDATA 1 , 2 , 3 ,");
    assert_eq!(eval_top(&mut vm, "MYDATA @"), 1);
    assert_eq!(eval_top(&mut vm, "MYDATA 1 + @"), 2);
    assert_eq!(eval_top(&mut vm, "MYDATA 2 + @"), 3);
}

#[test]
fn test_case_insensitive_lookup() {
    let mut vm = test_vm();
    // Built-in words
    assert_eq!(eval_top(&mut vm, "2 3 +"), 5);
    assert_eq!(eval_top(&mut vm, "2 dup +"), 4);
    assert_eq!(eval_top(&mut vm, "10 DUP *"), 100);
    // User-defined words
    eval(&mut vm, ": Square DUP * ;");
    assert_eq!(eval_top(&mut vm, "7 square"), 49);
    assert_eq!(eval_top(&mut vm, "7 SQUARE"), 49);
    assert_eq!(eval_top(&mut vm, "7 SqUaRe"), 49);
}

// -----------------------------------------------------------------------
// Swarm tests
// -----------------------------------------------------------------------

#[test]
fn test_swarm_status_word() {
    let mut vm = test_vm();
    let out = eval(&mut vm, "SWARM-STATUS");
    // Without mesh, shows offline
    assert!(out.contains("offline") || out.contains("swarm"));
}

#[test]
fn test_shared_words_empty() {
    let mut vm = test_vm();
    let out = eval(&mut vm, "SHARED-WORDS");
    // With no mesh, output is empty (graceful offline behavior).
    assert!(out.is_empty() || out.contains("no shared"));
}

#[test]
fn test_swarm_on_word() {
    let mut vm = test_vm();
    let out = eval(&mut vm, "SWARM-ON");
    assert!(out.contains("swarm mode"));
}

// -----------------------------------------------------------------------
// Signaling primitives (v0.28) — direct channel
// -----------------------------------------------------------------------

#[test]
fn test_say_pushes_to_outbox() {
    let mut vm = test_vm();
    let energy_before = vm.energy.energy;
    eval(&mut vm, "42 SAY!");
    assert_eq!(vm.outbox.len(), 1);
    assert_eq!(vm.outbox[0].value, 42);
    assert!(vm.outbox[0].is_direct());
    assert_eq!(
        vm.energy.energy,
        energy_before - crate::energy::SAY_COST,
        "SAY! must charge SAY_COST"
    );
    assert!(vm.stack.is_empty(), "SAY! consumes its argument");
}

#[test]
fn test_say_increments_signal_tick() {
    let mut vm = test_vm();
    eval(&mut vm, "1 SAY!");
    eval(&mut vm, "2 SAY!");
    eval(&mut vm, "3 SAY!");
    assert_eq!(vm.outbox.len(), 3);
    let ticks: Vec<u64> = vm.outbox.iter().map(|s| s.sent_at_tick).collect();
    assert!(ticks[0] < ticks[1] && ticks[1] < ticks[2]);
}

#[test]
fn test_say_no_op_when_starving() {
    let mut vm = test_vm();
    // Drain energy to just above the floor so we can't afford SAY_COST.
    vm.energy.energy = -498;
    let energy_before = vm.energy.energy;
    eval(&mut vm, "99 SAY!");
    assert!(vm.outbox.is_empty(), "starving unit must not emit");
    assert_eq!(vm.stack, vec![99], "no-op preserves stack");
    assert_eq!(vm.energy.energy, energy_before, "no-op charges nothing");
}

#[test]
fn test_listen_empty_pushes_zero() {
    let mut vm = test_vm();
    eval(&mut vm, "LISTEN");
    assert_eq!(vm.stack, vec![0]);
}

#[test]
fn test_listen_returns_oldest_value_minus_one() {
    let mut vm = test_vm();
    vm.inbox
        .push(crate::signaling::Signal::direct([0xaa; 8], 7, 1));
    vm.inbox
        .push(crate::signaling::Signal::direct([0xbb; 8], 11, 2));
    eval(&mut vm, "LISTEN");
    assert_eq!(vm.stack, vec![7, -1]);
    assert_eq!(vm.inbox.len(), 1);
}

#[test]
fn test_inbox_query_count() {
    let mut vm = test_vm();
    eval(&mut vm, "INBOX?");
    assert_eq!(vm.stack, vec![0]);
    vm.stack.clear();
    for i in 0..5 {
        vm.inbox
            .push(crate::signaling::Signal::direct([0; 8], i, i as u64));
    }
    eval(&mut vm, "INBOX?");
    assert_eq!(vm.stack, vec![5]);
}

#[test]
fn test_listen_does_not_charge_energy() {
    let mut vm = test_vm();
    vm.inbox
        .push(crate::signaling::Signal::direct([0; 8], 1, 0));
    let energy_before = vm.energy.energy;
    eval(&mut vm, "INBOX?");
    eval(&mut vm, "LISTEN");
    assert_eq!(vm.energy.energy, energy_before, "reads must be free");
}

// -----------------------------------------------------------------------
// Signaling primitives (v0.28) — environmental channel
// -----------------------------------------------------------------------

#[cfg(not(target_arch = "wasm32"))]
#[test]
fn test_mark_pushes_environmental_signal() {
    let mut vm = test_vm();
    // No dominant niche → should fall back to "general".
    let energy_before = vm.energy.energy;
    eval(&mut vm, "55 MARK!");
    assert_eq!(vm.outbox.len(), 1);
    let s = &vm.outbox[0];
    assert!(!s.is_direct(), "MARK! emits Environmental, not Direct");
    match &s.kind {
        crate::signaling::SignalKind::Environmental { niche } => {
            assert_eq!(niche, "general");
        }
        _ => panic!("expected Environmental kind"),
    }
    assert_eq!(s.value, 55);
    assert_eq!(
        vm.energy.energy,
        energy_before - crate::energy::MARK_COST,
        "MARK! must charge MARK_COST"
    );
}

#[cfg(not(target_arch = "wasm32"))]
#[test]
fn test_mark_uses_dominant_niche() {
    let mut vm = test_vm();
    vm.niche_profile
        .specializations
        .insert("fibonacci".to_string(), 0.9);
    eval(&mut vm, "100 MARK!");
    let s = &vm.outbox[0];
    match &s.kind {
        crate::signaling::SignalKind::Environmental { niche } => {
            assert_eq!(niche, "fibonacci");
        }
        _ => panic!("expected Environmental"),
    }
}

#[cfg(not(target_arch = "wasm32"))]
#[test]
fn test_mark_no_op_when_starving() {
    let mut vm = test_vm();
    vm.energy.energy = -498;
    let energy_before = vm.energy.energy;
    eval(&mut vm, "9 MARK!");
    assert!(vm.outbox.is_empty());
    assert_eq!(vm.stack, vec![9]);
    assert_eq!(vm.energy.energy, energy_before);
}

#[cfg(not(target_arch = "wasm32"))]
#[test]
fn test_sense_reads_env_view() {
    let mut vm = test_vm();
    vm.env_view = 42;
    let energy_before = vm.energy.energy;
    eval(&mut vm, "SENSE");
    assert_eq!(vm.stack, vec![42]);
    assert_eq!(vm.energy.energy, energy_before, "SENSE is free");
}

#[cfg(not(target_arch = "wasm32"))]
#[test]
fn test_sense_default_zero() {
    let mut vm = test_vm();
    eval(&mut vm, "SENSE");
    assert_eq!(vm.stack, vec![0]);
}

#[test]
fn test_court_prelude_word_emits_signal() {
    let mut vm = test_vm();
    // COURT should be defined by the prelude as `: COURT FITNESS SAY! ;`.
    // FITNESS reads the unit's fitness value; with no setup it returns 0.
    let energy_before = vm.energy.energy;
    eval(&mut vm, "COURT");
    assert_eq!(vm.outbox.len(), 1, "COURT should emit one signal");
    assert!(vm.outbox[0].is_direct());
    assert_eq!(
        vm.energy.energy,
        energy_before - crate::energy::SAY_COST,
        "COURT should charge SAY_COST via the SAY! it calls"
    );
}