bevy-brink 0.0.17

//! Tests for the ink→engine binding registration API (`super::registration`)
//! and the drive/eval API (`super::drive`) — most scenarios exercise both
//! together (register a binding, then drive a flow or an engine→ink call
//! through it), so the split module keeps a single shared test file.

use super::*;

use bevy_app::App;
use bevy_asset::Assets;
use bevy_ecs::prelude::*;
use brink_format::Value;
use brink_runtime::{ExternalFnHandler, ExternalResult, FastRng, FlowInstance};

use crate::BrinkFlow;
use crate::asset::{LineTablesAsset, ProgramAsset};
use crate::test_support::{compile_test_story, compile_test_story_brink_gradual};

/// A command event used by tests. `reply` is overridden to echo the
/// label length back to ink, exercising the value-return path.
#[derive(Event, Clone, Debug, PartialEq, Eq)]
struct Ping {
    label: String,
}

impl BrinkCommand for Ping {
    fn from_ink_args(args: &[Value]) -> Result<Self, BrinkArgError> {
        let label = args
            .first()
            .and_then(Value::as_str)
            .ok_or(BrinkArgError::Type {
                index: 0,
                expected: "string",
            })?
            .to_string();
        Ok(Self { label })
    }

    fn reply(&self) -> Value {
        #[expect(
            clippy::cast_possible_truncation,
            clippy::cast_possible_wrap,
            reason = "test value, small"
        )]
        Value::Int(self.label.len() as i32)
    }
}

/// A multi-field command whose `BrinkCommand` impl is generated by the
/// derive macro (strict types, default `reply` of `Null`).
#[derive(Event, Clone, Debug, PartialEq, bevy_brink_derive::BrinkCommand)]
struct SetVolume {
    channel: i32,
    level: f32,
}

fn app_with_double_and_ping() -> App {
    let mut app = App::new();
    app.bind_brink_fn::<(), _, _>("double", |args| {
        args.first().and_then(Value::as_int).unwrap_or(0) * 2
    });
    app.bind_brink_command::<(), Ping>("ping");
    app
}

#[test]
fn pure_fn_resolves_inline() {
    let app = app_with_double_and_ping();
    let bindings = app.world().resource::<BrinkBindings<()>>();
    let handler = bindings.handler();
    match handler.call("double", &[Value::Int(21)]) {
        ExternalResult::Resolved(Value::Int(42)) => {}
        other => panic!("expected Resolved(Int(42)), got {other:?}"),
    }
    assert_eq!(handler.queued_len(), 0, "pure fn buffers nothing");
}

#[test]
fn command_buffers_trigger_and_returns_reply() {
    let app = app_with_double_and_ping();
    let bindings = app.world().resource::<BrinkBindings<()>>();
    let handler = bindings.handler();
    // "hi" has length 2 → reply Int(2); one trigger buffered.
    match handler.call("ping", &[Value::from("hi")]) {
        ExternalResult::Resolved(Value::Int(2)) => {}
        other => panic!("expected Resolved(Int(2)), got {other:?}"),
    }
    assert_eq!(handler.queued_len(), 1, "command buffers one trigger");
}

#[test]
fn unknown_name_falls_back() {
    let app = app_with_double_and_ping();
    let bindings = app.world().resource::<BrinkBindings<()>>();
    let handler = bindings.handler();
    match handler.call("nonexistent", &[]) {
        ExternalResult::Fallback => {}
        other => panic!("expected Fallback, got {other:?}"),
    }
}

#[test]
fn bad_command_args_resolve_null_without_buffering() {
    let app = app_with_double_and_ping();
    let bindings = app.world().resource::<BrinkBindings<()>>();
    let handler = bindings.handler();
    // ping wants a string; give it an int → parse fails → Null, no buffer.
    match handler.call("ping", &[Value::Int(7)]) {
        ExternalResult::Resolved(Value::Null) => {}
        other => panic!("expected Resolved(Null), got {other:?}"),
    }
    assert_eq!(handler.queued_len(), 0, "failed parse buffers nothing");
}

/// The snapshot-only contract (value-model-spec §8/§10): a binding receives
/// argument values by shared reference. Cloning one is an O(1) Arc snapshot,
/// and any mutation forks via copy-on-write, so a binding can never mutate
/// script state through an argument — the host holds a frozen snapshot, not
/// a live handle. This locks that guarantee at the bevy-brink binding
/// surface for collection values specifically.
#[test]
fn binding_arg_is_a_snapshot_not_a_mutable_handle() {
    let mut app = App::new();
    // Clone the incoming array (a snapshot), mutate the clone, and report
    // its new length. The caller's array must be untouched.
    app.bind_brink_fn::<(), _, _>("append_len", |args| {
        let mut snap = args.first().cloned().unwrap_or(Value::Null);
        match snap.array_make_mut() {
            Some(items) => {
                items.push(Value::Int(99));
                #[expect(clippy::cast_possible_truncation, clippy::cast_possible_wrap)]
                Value::Int(items.len() as i32)
            }
            None => Value::Int(-1),
        }
    });
    let bindings = app.world().resource::<BrinkBindings<()>>();
    let handler = bindings.handler();

    let original = Value::array(vec![Value::Int(1), Value::Int(2)]);
    // Pass `original` itself (by shared ref) — not a pre-made clone — so the
    // only copy the binding can obtain is the snapshot it clones internally.
    match handler.call("append_len", std::slice::from_ref(&original)) {
        ExternalResult::Resolved(Value::Int(3)) => {}
        other => panic!("expected Resolved(Int(3)), got {other:?}"),
    }
    assert_eq!(
        original.as_array().expect("array").as_slice(),
        &[Value::Int(1), Value::Int(2)],
        "binding mutated only its snapshot; the caller's value is unchanged",
    );
}

/// A collection round-trips through a binding return unchanged — the
/// `R: Into<Value>` return path carries `Array`/`Map` values (a snapshot
/// handed back to the script), not just scalars.
#[test]
fn binding_can_return_a_collection_snapshot() {
    use brink_format::{MapKey, OrderedMap};

    let mut app = App::new();
    app.bind_brink_fn::<(), _, _>("make_map", |_args| {
        let m: OrderedMap = [
            (MapKey::from("a"), Value::Int(1)),
            (MapKey::from("b"), Value::Int(2)),
        ]
        .into_iter()
        .collect();
        Value::map(m)
    });
    let bindings = app.world().resource::<BrinkBindings<()>>();
    let handler = bindings.handler();
    match handler.call("make_map", &[]) {
        ExternalResult::Resolved(v) => {
            let map = v.as_map().expect("returned a map");
            assert_eq!(map.len(), 2);
            assert_eq!(map.get(&MapKey::from("a")), Some(&Value::Int(1)));
        }
        other => panic!("expected Resolved(map), got {other:?}"),
    }
}

// ── T1e pass-through audit (docs/t1e-spec.md §8 item 3, issue #850) ──
//
// "bevy-brink pass-through audit: projections cross bindings as
// ordinary values; the host never walks paths — assert it." Two halves:
//
// 1. A raw `Value::Projection` can *structurally never* reach a binding
//    from real compiled ink. `ref lvalue-path` (`ref npc.hp`) only
//    lowers to `MakeProjection` when the target's own declared
//    parameter is `ref` (`brink_ir::lir::lower::expr::lower_call_args`:
//    `is_ref = params[i].is_ref`) — and an `EXTERNAL` declaration's
//    grammar has no `ref` marker at all
//    (`brink_syntax::parser::declaration::function_param_list` parses
//    plain identifiers only, unlike a knot/function header's
//    `KNOT_PARAMS`, which does). So `heal(ref npc.hp, 5)` against an
//    `EXTERNAL heal` can never lower to a projection; it hits the
//    ordinary ref-argument fence and fails to compile. `bindings.rs`'s
//    `ExternalFnHandler::call` therefore never sees the `(root,
//    segments)` descriptor — only a real, structural guarantee.
// 2. Once a projection-bound `ref` parameter *is* read inside an
//    ordinary ink function body (the only way to observe one), the
//    read auto-dereferences to a plain value before it can be used as
//    an argument anywhere (`vm.rs`'s `GetTemp` dispatch, T1e-2) — so a
//    value derived from a projection and handed to an external binding
//    always arrives as an ordinary snapshot, never the projection
//    itself.

/// Half 1: `ref npc.hp` against an `EXTERNAL` target fails to compile —
/// there is no lowering path that could hand a raw `Value::Projection`
/// to a host binding.
#[test]
fn ref_projection_argument_to_an_external_call_fails_to_compile() {
    use brink_compiler::{AnalysisOptions, CompileError, Dialect};
    let src = "EXTERNAL heal(target, amount)\n\
                    STRUCT NPC = #{hp: int}\n\
                    VAR npc = 0\n\
                    === main ===\n\
                    ~ npc = NPC#{hp: 1}\n\
                    ~ heal(ref npc.hp, 5)\n\
                    -> END\n";
    let err = brink_compiler::compile_with_options(
        "audit.ink",
        |path| {
            if path == "audit.ink" {
                Ok(src.to_string())
            } else {
                Err(std::io::Error::new(std::io::ErrorKind::NotFound, path))
            }
        },
        AnalysisOptions {
            dialect: Dialect::Brink,
            ..AnalysisOptions::default()
        },
    )
    .expect_err("a ref-projection argument to an EXTERNAL target must not compile");
    let CompileError::Diagnostics(diags) = err else {
        panic!("expected a diagnostics rejection, got {err:?}");
    };
    assert!(
        !diags.is_empty(),
        "expected at least one diagnostic rejecting the construct"
    );
}

/// Half 2: an ink function with a projection-bound `ref` parameter
/// reads it and forwards the *resolved* value to an external binding —
/// the binding observes a plain `Value::Int`, never a `Value::Projection`.
#[test]
fn value_read_through_a_ref_projection_reaches_a_binding_as_a_plain_snapshot() {
    let src = "STRUCT NPC = #{hp: int, name: string}\n\
                    VAR npc = 0\n\
                    EXTERNAL report(x)\n\n\
                    ~ npc = NPC#{hp: 7, name: \"x\"}\n\
                    ~ temp result = heal(ref npc.hp, 5)\n\
                    Result: {result}\n\
                    -> END\n\n\
                    === function heal(ref hp, amount) ===\n\
                    ~ temp seen = report(hp)\n\
                    ~ hp = hp + amount\n\
                    ~ return hp\n";
    // NS-A9 (strict brink default): the `VAR npc = 0` → struct-reassign
    // placeholder idiom predates #1530, which made a well-formed
    // construction literal a legal declaration default; the idiom is still
    // gradual-locked (strict types `npc` as the scalar and rejects at the
    // reassignment), and the subject here (projection resolution at the
    // binding seam) is regime-independent. Migrating this fixture onto the
    // direct spelling is a separate pass.
    let (program, tables, _ctx) = compile_test_story_brink_gradual(src);

    let seen: std::sync::Arc<std::sync::Mutex<Vec<Value>>> = std::sync::Arc::default();
    let seen_for_binding = std::sync::Arc::clone(&seen);
    let mut app = App::new();
    app.bind_brink_fn::<(), _, _>("report", move |args| {
        seen_for_binding
            .lock()
            .expect("lock")
            .push(args.first().cloned().unwrap_or(Value::Null));
        Value::Null
    });
    let bindings = app.world().resource::<BrinkBindings<()>>();
    let handler = bindings.handler();

    let (mut flow, mut ctx) = FlowInstance::new_at_root(&program);
    let mut text = String::new();
    loop {
        let line = flow
            .step_single_line::<FastRng>(&program, &tables, &mut ctx, &handler, None)
            .unwrap();
        text.push_str(line.text());
        if line.is_terminal() {
            break;
        }
    }

    assert!(
        text.contains("Result: 12"),
        "heal(ref npc.hp, 5) should read 7, add 5, and return 12; got {text:?}"
    );
    let captured = seen.lock().expect("lock");
    assert_eq!(
        captured.as_slice(),
        &[Value::Int(7)],
        "the binding must see the dereferenced value (7), never a Value::Projection: \
             got {captured:?}"
    );
}

/// End-to-end: a pure-fn binding's return value is inlined into story
/// text by the real VM.
#[test]
fn e2e_pure_fn_value_appears_in_text() {
    let (program, tables, _ctx) =
        compile_test_story("EXTERNAL double(x)\nResult: {double(21)}.\n-> END\n");

    let app = app_with_double_and_ping();
    let bindings = app.world().resource::<BrinkBindings<()>>();
    let handler = bindings.handler();

    let (mut flow, mut ctx) = FlowInstance::new_at_root(&program);
    let mut text = String::new();
    loop {
        let line = flow
            .step_single_line::<FastRng>(&program, &tables, &mut ctx, &handler, None)
            .unwrap();
        text.push_str(line.text());
        if line.is_terminal() {
            break;
        }
    }
    assert!(
        text.contains("Result: 42"),
        "expected 'Result: 42' in story text; got {text:?}"
    );
}

/// End-to-end: a command binding fires its Bevy event when the VM hits
/// the external call. We drive the VM, then apply the buffered trigger
/// to the world and confirm an observer saw the event.
#[test]
fn e2e_command_triggers_event() {
    #[derive(Resource, Default)]
    struct PingLog(Vec<String>);

    let (program, tables, _ctx) =
        compile_test_story("EXTERNAL ping(label)\nA{ping(\"hi\")}B\n-> END\n");

    let mut app = app_with_double_and_ping();
    app.init_resource::<PingLog>();
    app.add_observer(|on: On<Ping>, mut log: ResMut<PingLog>| {
        log.0.push(on.event().label.clone());
    });

    // Drive the flow inside a scope so the borrow of BrinkBindings
    // ends before we mutate the world to apply triggers.
    let triggers = {
        let bindings = app.world().resource::<BrinkBindings<()>>();
        let handler = bindings.handler();
        let (mut flow, mut ctx) = FlowInstance::new_at_root(&program);
        loop {
            let line = flow
                .step_single_line::<FastRng>(&program, &tables, &mut ctx, &handler, None)
                .unwrap();
            if line.is_terminal() {
                break;
            }
        }
        handler.queued.into_inner()
    };
    assert_eq!(triggers.len(), 1, "exactly one ping trigger buffered");

    for trigger in triggers {
        trigger(app.world_mut());
    }

    let log = app.world().resource::<PingLog>();
    assert_eq!(
        log.0,
        vec!["hi".to_string()],
        "observer should see ping(\"hi\")"
    );
}

#[test]
fn derived_from_ink_args_parses_strictly() {
    // Correct count + types.
    let ok = SetVolume::from_ink_args(&[Value::Int(2), Value::Float(0.5)]).unwrap();
    assert_eq!(
        ok,
        SetVolume {
            channel: 2,
            level: 0.5
        }
    );
    // Default reply is Null.
    assert!(matches!(ok.reply(), Value::Null));

    // Wrong count.
    assert_eq!(
        SetVolume::from_ink_args(&[Value::Int(2)]),
        Err(BrinkArgError::Count {
            expected: 2,
            got: 1
        })
    );

    // Wrong type at index 1 (int where float expected — strict, no
    // coercion, mirroring bladeink's derive).
    assert_eq!(
        SetVolume::from_ink_args(&[Value::Int(2), Value::Int(3)]),
        Err(BrinkArgError::Type {
            index: 1,
            expected: "float"
        })
    );
}

/// End-to-end: a derived command binding fires its event when the VM
/// hits the external call, just like a hand-written one.
#[test]
fn e2e_derived_command_triggers_event() {
    #[derive(Resource, Default)]
    struct VolumeLog(Vec<(i32, f32)>);

    let (program, tables, _ctx) =
        compile_test_story("EXTERNAL set_volume(ch, lvl)\nA{set_volume(2, 0.5)}B\n-> END\n");

    let mut app = App::new();
    app.bind_brink_command::<(), SetVolume>("set_volume");
    app.init_resource::<VolumeLog>();
    app.add_observer(|on: On<SetVolume>, mut log: ResMut<VolumeLog>| {
        log.0.push((on.event().channel, on.event().level));
    });

    let triggers = {
        let bindings = app.world().resource::<BrinkBindings<()>>();
        let handler = bindings.handler();
        let (mut flow, mut ctx) = FlowInstance::new_at_root(&program);
        loop {
            let line = flow
                .step_single_line::<FastRng>(&program, &tables, &mut ctx, &handler, None)
                .unwrap();
            if line.is_terminal() {
                break;
            }
        }
        handler.queued.into_inner()
    };
    assert_eq!(triggers.len(), 1);
    for trigger in triggers {
        trigger(app.world_mut());
    }

    let log = app.world().resource::<VolumeLog>();
    assert_eq!(
        log.0,
        vec![(2, 0.5)],
        "observer should see set_volume(2, 0.5)"
    );
}

// ── Engine → ink: call_ink_function + bind_brink_query ───────────

#[derive(Component)]
struct Enemy;

/// A world-access query binding: count `Enemy` entities.
fn enemy_count(In((_entity, _args)): In<BrinkQueryInput>, enemies: Query<&Enemy>) -> Value {
    #[expect(clippy::cast_possible_truncation, clippy::cast_possible_wrap)]
    Value::Int(enemies.iter().count() as i32)
}

/// End-to-end engine→ink: an ink function that queries the World via a
/// `bind_brink_query` binding, driven synchronously by
/// `call_ink_function`.
#[test]
fn call_ink_function_resolves_world_query() {
    use crate::BrinkFlowRequest;
    use crate::test_support::{add_story_assets, make_test_app};

    let mut app = make_test_app();
    app.bind_brink_query::<(), _, _>("enemy_count", enemy_count);

    // can_spawn() := enemy_count() < 3
    let (program, tables, ctx) = compile_test_story(
        "EXTERNAL enemy_count()\n-> END\n=== function can_spawn() ===\n~ return enemy_count() < 3\n",
    );
    let story = add_story_assets(&mut app, program, tables, ctx);

    // Two enemies → can_spawn should be true (2 < 3).
    app.world_mut().spawn(Enemy);
    app.world_mut().spawn(Enemy);

    let entity = app
        .world_mut()
        .spawn(BrinkFlowRequest::<()>::builder().story(story).build())
        .id();
    app.update(); // fulfill the request → flow components on `entity`

    let result = call_ink_function::<()>(app.world_mut(), entity, "can_spawn", &[]).unwrap();
    assert_eq!(result.as_bool(), Some(true), "2 enemies < 3 → can spawn");

    // Add two more enemies (4 total) → can_spawn should now be false.
    app.world_mut().spawn(Enemy);
    app.world_mut().spawn(Enemy);
    let result = call_ink_function::<()>(app.world_mut(), entity, "can_spawn", &[]).unwrap();
    assert_eq!(
        result.as_bool(),
        Some(false),
        "4 enemies !< 3 → cannot spawn"
    );
}

/// End-to-end host callback-invocation surface (T1c-3, `docs/t1c-spec.md`
/// §6): an ink function returns a **function value** as an opaque token to
/// the host, which then re-invokes it via `call_ink_function_value` —
/// re-entering the VM without ever dereferencing the token's env. Both a
/// zero-bound `FnRef` and a `bind`-curried `Closure` are exercised.
#[test]
fn call_ink_function_value_reinvokes_an_opaque_token() {
    use crate::BrinkFlowRequest;
    use crate::test_support::{add_story_assets, make_test_app};

    let mut app = make_test_app();
    // Register a (never-called) binding so the `BrinkBindings<()>` resource
    // the engine→ink driver reads exists — this fixture uses no externals.
    app.bind_brink_fn::<(), _, _>("_unused", |_| 0);

    // `make_doubler` returns a bare `#fn(double)`; `make_adder` returns a
    // `bind`-curried closure that has bound the first param.
    let (program, tables, ctx) = crate::test_support::compile_test_story_brink(
        "-> END\n\
             === function make_doubler() ===\n~ return #fn(double)\n\
             === function make_adder() ===\n~ return bind(#fn(add), 10)\n\
             === function double(x: int): int ===\n~ return x + x\n\
             === function add(a: int, b: int): int ===\n~ return a + b\n",
    );
    let story = add_story_assets(&mut app, program, tables, ctx);
    let entity = app
        .world_mut()
        .spawn(BrinkFlowRequest::<()>::builder().story(story).build())
        .id();
    app.update();

    // Obtain the opaque function-value tokens from ink.
    let double_token =
        call_ink_function::<()>(app.world_mut(), entity, "make_doubler", &[]).unwrap();
    let adder_token = call_ink_function::<()>(app.world_mut(), entity, "make_adder", &[]).unwrap();

    // Re-invoke each through the host callback surface. The host never
    // inspects the token — invocation re-enters the VM.
    let doubled =
        call_ink_function_value::<()>(app.world_mut(), entity, &double_token, &[Value::Int(21)])
            .unwrap();
    assert_eq!(doubled.as_int(), Some(42), "double(21)");

    let summed =
        call_ink_function_value::<()>(app.world_mut(), entity, &adder_token, &[Value::Int(5)])
            .unwrap();
    assert_eq!(summed.as_int(), Some(15), "add(10, 5) via bound closure");

    // Wrong arity through the opaque token is a defined fault, not silent
    // garbage (spec §3), surfaced as a `Runtime` error.
    let err = call_ink_function_value::<()>(
        app.world_mut(),
        entity,
        &double_token,
        &[Value::Int(1), Value::Int(2)],
    )
    .unwrap_err();
    assert!(
        matches!(err, BrinkCallError::Runtime(_)),
        "over-supplied args must fault, got {err:?}"
    );
}

/// A pure binding called from inside an engine→ink function resolves
/// inline (no World access), and an unknown function errors clearly.
#[test]
fn call_ink_function_pure_and_errors() {
    use crate::BrinkFlowRequest;
    use crate::test_support::{add_story_assets, make_test_app};

    let mut app = make_test_app();
    app.bind_brink_fn::<(), _, _>("triple", |args| {
        args.first().and_then(Value::as_int).unwrap_or(0) * 3
    });

    let (program, tables, ctx) = compile_test_story(
        "EXTERNAL triple(n)\n-> END\n=== function scaled(n) ===\n~ return triple(n) + 1\n",
    );
    let story = add_story_assets(&mut app, program, tables, ctx);
    let entity = app
        .world_mut()
        .spawn(BrinkFlowRequest::<()>::builder().story(story).build())
        .id();
    app.update();

    // triple(7) + 1 = 22
    let result =
        call_ink_function::<()>(app.world_mut(), entity, "scaled", &[Value::Int(7)]).unwrap();
    assert_eq!(result, Value::Int(22));

    // Unknown function → clear error.
    let err = call_ink_function::<()>(app.world_mut(), entity, "nope", &[]).unwrap_err();
    assert!(
        matches!(err, BrinkCallError::FunctionNotFound(_)),
        "got {err:?}"
    );
}

/// Regression for issue #1096: a `bind_brink_command`-bound `EXTERNAL`
/// reached through `call_ink_function` (the exclusive engine→ink eval
/// driver) must fire its Bevy event, not silently run the in-story
/// fallback the way an *unbound* external would.
///
/// The fixture gives `ping` an in-story fallback (`~ return -1`) so the
/// pre-fix bug is genuinely *silent*: `EvalHandler::call` used to fall
/// through to `ExternalResult::Fallback` for a command binding exactly as
/// it does for an unbound name, so the call would resolve `Ok(-1)` with no
/// error and no event — indistinguishable from a legitimate call unless
/// something asserts the event actually fired. Asserting only that the
/// call returns `Ok` (as `call_ink_function_pure_and_errors` does for pure
/// bindings) would therefore pass on both the broken and fixed code; this
/// test asserts both the command's real reply (not the fallback's `-1`)
/// *and* the observer log, so it fails loudly against the silent-fallback
/// bug.
#[test]
fn call_ink_function_fires_command_event_not_silent_fallback() {
    use crate::BrinkFlowRequest;
    use crate::test_support::{add_story_assets, make_test_app};

    #[derive(Resource, Default)]
    struct PingLog(Vec<String>);

    let mut app = make_test_app();
    app.bind_brink_command::<(), Ping>("ping");
    app.init_resource::<PingLog>();
    app.add_observer(|on: On<Ping>, mut log: ResMut<PingLog>| {
        log.0.push(on.event().label.clone());
    });

    let (program, tables, ctx) = compile_test_story(
        "EXTERNAL ping(label)\n\
         -> END\n\
         === function ping(label) ===\n\
         ~ return -1\n\
         === function sound_alarm() ===\n\
         ~ return ping(\"intruder\")\n",
    );
    let story = add_story_assets(&mut app, program, tables, ctx);
    let entity = app
        .world_mut()
        .spawn(BrinkFlowRequest::<()>::builder().story(story).build())
        .id();
    app.update();

    let result = call_ink_function::<()>(app.world_mut(), entity, "sound_alarm", &[]).unwrap();

    // `Ping::reply` returns the label's length (`"intruder".len() == 8`);
    // the in-story fallback would have returned `-1`. A `-1` here would
    // mean the call quietly ran the fallback instead of the binding.
    assert_eq!(
        result,
        Value::Int(8),
        "expected the command binding's reply (label length 8), not the \
             in-story fallback's -1 — the command binding must resolve the call"
    );

    // The actual regression: the event must have fired, not merely have
    // the call return a plausible value.
    let log = app.world().resource::<PingLog>();
    assert_eq!(
        log.0,
        vec!["intruder".to_string()],
        "call_ink_function must fire the bind_brink_command event, not \
             silently fall back to the in-story body; observer log: {:?}",
        log.0
    );
}

/// Regression for issue #1096, the [`call_ink_function_value`] sibling of
/// [`call_ink_function_fires_command_event_not_silent_fallback`]: a
/// `bind_brink_command`-bound `EXTERNAL` reached through the **function
/// value** re-entry path (its own `EvalHandler`/`flush_eval_triggers` wiring
/// in `drive_function_eval_to_done`/`call_ink_function_value`, separate from
/// `call_ink_function`'s) must also fire its Bevy event, not silently run
/// the in-story fallback.
///
/// The obtained token points at `sound_alarm`, which itself calls the
/// `ping` external — so invoking the token re-enters the VM and hits
/// `EvalHandler::call` on the command binding exactly as a by-name call
/// would, but through `call_ink_function_value`'s independent driver code.
/// As with the by-name regression test, asserting only that the call
/// returns `Ok` would pass on both the broken and fixed code (the command
/// binding's `Resolved` reply doesn't depend on whether its trigger is ever
/// flushed against the World) — the observer log is what actually catches a
/// missing `flush_eval_triggers` call on this path.
#[test]
fn call_ink_function_value_fires_command_event_not_silent_fallback() {
    use crate::BrinkFlowRequest;
    use crate::test_support::{add_story_assets, make_test_app};

    #[derive(Resource, Default)]
    struct PingLog(Vec<String>);

    let mut app = make_test_app();
    app.bind_brink_command::<(), Ping>("ping");
    app.init_resource::<PingLog>();
    app.add_observer(|on: On<Ping>, mut log: ResMut<PingLog>| {
        log.0.push(on.event().label.clone());
    });

    // Needs the brink dialect for `#fn(…)` (function values); `_gradual`
    // opts out of NS-A9's strict-by-default typing since the subject here
    // (command bindings reached via a function-value token) is
    // regime-independent — see `compile_test_story_brink_gradual`'s doc.
    let (program, tables, ctx) = crate::test_support::compile_test_story_brink_gradual(
        "EXTERNAL ping(label)\n\
         -> END\n\
         === function ping(label) ===\n\
         ~ return -1\n\
         === function sound_alarm() ===\n\
         ~ return ping(\"intruder\")\n\
         === function make_alarm_caller() ===\n\
         ~ return #fn(sound_alarm)\n",
    );
    let story = add_story_assets(&mut app, program, tables, ctx);
    let entity = app
        .world_mut()
        .spawn(BrinkFlowRequest::<()>::builder().story(story).build())
        .id();
    app.update();

    let token = call_ink_function::<()>(app.world_mut(), entity, "make_alarm_caller", &[]).unwrap();
    let result = call_ink_function_value::<()>(app.world_mut(), entity, &token, &[]).unwrap();

    // `Ping::reply` returns the label's length (8); the in-story fallback
    // would have returned -1.
    assert_eq!(
        result,
        Value::Int(8),
        "expected the command binding's reply (label length 8) via the function-value \
             re-entry path, not the in-story fallback's -1"
    );

    // The actual regression: the event must have fired through this path's
    // own EvalHandler/flush_eval_triggers wiring.
    let log = app.world().resource::<PingLog>();
    assert_eq!(
        log.0,
        vec!["intruder".to_string()],
        "call_ink_function_value must fire the bind_brink_command event, not \
             silently fall back to the in-story body; observer log: {:?}",
        log.0
    );
}

/// A story line that calls a world-access query binding inline
/// (`{enemy_count()}`) resolves transparently when driven by
/// `advance_flow`.
#[test]
fn advance_flow_resolves_inline_query_during_playback() {
    use crate::BrinkFlowRequest;
    use crate::test_support::{add_story_assets, make_test_app};

    let mut app = make_test_app();
    app.bind_brink_query::<(), _, _>("enemy_count", enemy_count);

    let (program, tables, ctx) =
        compile_test_story("EXTERNAL enemy_count()\nEnemies near: {enemy_count()}.\n-> END\n");
    let story = add_story_assets(&mut app, program, tables, ctx);
    app.world_mut().spawn(Enemy);
    app.world_mut().spawn(Enemy);

    let entity = app
        .world_mut()
        .spawn(BrinkFlowRequest::<()>::builder().story(story).build())
        .id();
    app.update(); // fulfill

    let line = advance_flow::<()>(app.world_mut(), entity).unwrap();
    assert!(
        line.text().contains("Enemies near: 2."),
        "inline query should resolve to 2; got {:?}",
        line.text()
    );
}

/// The non-exclusive path: a normal `step_one` driver pauses on a query
/// (`Advance::AwaitingQuery`), the plugin's resolver resolves it across
/// frames, and the driver resumes to produce the line.
#[test]
#[expect(clippy::type_complexity, reason = "bevy driver closure query tuple")]
fn step_one_query_pauses_and_plugin_resolver_resumes() {
    use crate::test_support::{add_story_assets, make_test_app};
    use crate::{Advance, BrinkContext, BrinkFlow, BrinkFlowRequest, BrinkLocale, BrinkProgram};
    use bevy_app::Update;

    #[derive(Resource, Default)]
    struct Lines(Vec<String>);

    let mut app = make_test_app();
    app.init_resource::<Lines>();
    app.bind_brink_query::<(), _, _>("enemy_count", enemy_count);

    let (program, tables, ctx) =
        compile_test_story("EXTERNAL enemy_count()\nEnemies near: {enemy_count()}.\n-> END\n");
    let story = add_story_assets(&mut app, program, tables, ctx);
    app.world_mut().spawn(Enemy);
    app.world_mut().spawn(Enemy);
    app.world_mut()
        .spawn(BrinkFlowRequest::<()>::builder().story(story).build());

    // A normal (non-exclusive) driver: step each flow once per frame,
    // skipping flows paused on a query (the plugin resolver handles
    // those; we resume next frame).
    app.add_systems(
        Update,
        |mut flows: Query<(
            Entity,
            &mut BrinkFlow<()>,
            &mut BrinkContext<()>,
            &BrinkProgram<()>,
            &BrinkLocale<()>,
        )>,
         globals: Option<ResMut<crate::BrinkGlobals<()>>>,
         programs: Res<Assets<ProgramAsset>>,
         tables: Res<Assets<LineTablesAsset>>,
         bindings: Res<BrinkBindings<()>>,
         mut commands: Commands,
         mut out: ResMut<Lines>| {
            let Some(mut globals) = globals else {
                return;
            };
            for (entity, mut flow, mut ctx, prog, loc) in &mut flows {
                if flow.inner.has_pending_external() {
                    continue; // paused on a query; wait for the resolver
                }
                let (Some(p), Some(t)) = (programs.get(&prog.handle), tables.get(&loc.handle))
                else {
                    continue;
                };
                let handler = bindings.handler();
                let mut view = crate::globals::flow_context_view(&mut globals, &mut ctx);
                if let Ok(Advance::Step(line)) = flow.step_one(
                    &p.program,
                    &t.tables,
                    &mut view,
                    &handler,
                    entity,
                    &mut commands,
                ) {
                    out.0.push(line.text().to_string());
                }
                handler.flush(&mut commands);
            }
        },
    );

    // First update fulfills the request; subsequent updates drive +
    // resolve. A handful is plenty regardless of intra-frame ordering.
    for _ in 0..6 {
        app.update();
    }

    let lines = &app.world().resource::<Lines>().0;
    assert!(
        lines.iter().any(|l| l.contains("Enemies near: 2.")),
        "expected the resolved inline-query line; got {lines:?}"
    );
}

// ── Batch engine → ink: call_ink_functions (BH-5, #1058) ─────────

/// A stateful accumulator story: `add(n)` mutates a global and returns the
/// running total, `get()` reads it. Used by the batch tests to prove
/// ordering and cross-call state visibility.
fn compile_accumulator_story() -> (
    brink_runtime::Program,
    Vec<Vec<brink_format::LineEntry>>,
    brink_runtime::World,
) {
    crate::test_support::compile_test_story(
        "VAR total = 0\n-> END\n\
             === function add(n) ===\n~ total = total + n\n~ return total\n\
             === function get() ===\n~ return total\n",
    )
}

/// Build a fresh app with the accumulator story fulfilled, returning the
/// app and its flow entity. Each app owns its own `BrinkGlobals<()>`
/// resource, so two of them don't share the `total` global — the isolation
/// the parity comparison needs.
fn accumulator_app() -> (bevy_app::App, Entity) {
    use crate::BrinkFlowRequest;
    use crate::test_support::{add_story_assets, make_test_app};

    let mut app = make_test_app();
    let (p, t, c) = compile_accumulator_story();
    let story = add_story_assets(&mut app, p, t, c);
    app.world_mut()
        .spawn(BrinkFlowRequest::<()>::builder().story(story).build());
    app.update();
    let flow = app
        .world_mut()
        .query_filtered::<Entity, With<BrinkFlow<()>>>()
        .iter(app.world())
        .next()
        .expect("flow fulfilled");
    (app, flow)
}

/// **Parity property:** an N-call batch is byte-identical to N standalone
/// [`call_ink_function`]s applied to an identical flow, in the same order.
/// This is the correctness bar for the amortized entry point. The two paths
/// run in separate apps because story globals (`total`) are a per-marker
/// resource shared by every flow of that marker in one World.
#[test]
fn batch_equals_n_single_calls() {
    let calls: Vec<(&str, Vec<Value>)> = vec![
        ("add", vec![Value::Int(1)]),
        ("add", vec![Value::Int(10)]),
        ("add", vec![Value::Int(100)]),
        ("get", vec![]),
    ];

    // Batch path: one VM-eval setup for all four calls.
    let (mut batch_app, batch_flow) = accumulator_app();
    let batch: Vec<Value> =
        call_ink_functions::<(), _, _>(batch_app.world_mut(), batch_flow, calls.clone())
            .into_iter()
            .map(|r| r.unwrap())
            .collect();

    // Single-call path: four standalone setups on a twin flow in its own
    // World (its own globals).
    let (mut single_app, single_flow) = accumulator_app();
    let single: Vec<Value> = calls
        .iter()
        .map(|(name, args)| {
            call_ink_function::<()>(single_app.world_mut(), single_flow, name, args).unwrap()
        })
        .collect();

    assert_eq!(batch, single, "batch must equal N single calls");
    // And the concrete accumulation: 1, 11, 111, then get() == 111.
    assert_eq!(
        batch,
        vec![
            Value::Int(1),
            Value::Int(11),
            Value::Int(111),
            Value::Int(111)
        ],
        "ordered accumulation across the batch"
    );
}

/// **Error isolation + no silent drops:** a failing call yields `Err` in
/// *its* slot without aborting the batch; every later call still runs, and
/// the failed call leaves no partial state behind.
#[test]
fn batch_isolates_errors_and_preserves_order() {
    use crate::BrinkFlowRequest;
    use crate::test_support::{add_story_assets, make_test_app};

    let mut app = make_test_app();
    let (p, t, c) = compile_accumulator_story();
    let story = add_story_assets(&mut app, p, t, c);
    let flow = app
        .world_mut()
        .spawn(BrinkFlowRequest::<()>::builder().story(story).build())
        .id();
    app.update();

    let calls: Vec<(&str, Vec<Value>)> = vec![
        ("add", vec![Value::Int(1)]),
        ("nope", vec![]), // unknown function — must not abort the batch
        ("add", vec![Value::Int(10)]),
        ("get", vec![]),
    ];
    let results = call_ink_functions::<(), _, _>(app.world_mut(), flow, calls);

    assert_eq!(results.len(), 4, "one slot per call, no drops");
    assert_eq!(results[0].as_ref().unwrap(), &Value::Int(1));
    assert!(
        matches!(results[1], Err(BrinkCallError::FunctionNotFound(_))),
        "the bad call fails in its own slot; got {:?}",
        results[1]
    );
    // The failed call did not perturb the accumulator: the next add sees 1.
    assert_eq!(results[2].as_ref().unwrap(), &Value::Int(11));
    assert_eq!(results[3].as_ref().unwrap(), &Value::Int(11));
}

/// A batch whose calls each go through a **world-access query** binding
/// (`run_system_with` between suspensions) reuses the single `SystemState`
/// across every call — proving the re-borrow dance is safe batch-wide.
#[test]
fn batch_drives_world_queries_across_calls() {
    use crate::BrinkFlowRequest;
    use crate::test_support::{add_story_assets, make_test_app};

    let mut app = make_test_app();
    app.bind_brink_query::<(), _, _>("enemy_count", enemy_count);
    let (program, tables, ctx) = compile_test_story(
        "EXTERNAL enemy_count()\n-> END\n\
             === function seen() ===\n~ return enemy_count()\n",
    );
    let story = add_story_assets(&mut app, program, tables, ctx);
    app.world_mut().spawn(Enemy);
    app.world_mut().spawn(Enemy);
    let flow = app
        .world_mut()
        .spawn(BrinkFlowRequest::<()>::builder().story(story).build())
        .id();
    app.update();

    let calls: Vec<(&str, Vec<Value>)> = vec![("seen", vec![]), ("seen", vec![]), ("seen", vec![])];
    let results: Vec<Value> = call_ink_functions::<(), _, _>(app.world_mut(), flow, calls)
        .into_iter()
        .map(|r| r.unwrap())
        .collect();

    assert_eq!(
        results,
        vec![Value::Int(2), Value::Int(2), Value::Int(2)],
        "each batched query resolves against the World"
    );
}

/// Regression proving `call_ink_functions`' documented **per-call** flush
/// contract (`drive.rs`'s `flush_eval_triggers` call sitting inside the
/// batch's per-call loop, not once after the whole batch): call 1 fires a
/// `bind_brink_command` whose observer writes a resource; call 2 is a
/// `bind_brink_query` that reads it. Only per-call flush lets call 2 observe
/// call 1's World effect — an implementation that buffered every call's
/// triggers and flushed once at the end of the batch would still resolve
/// both calls (`Resolved` doesn't depend on flush timing), so this can
/// *only* pass under the isolated-per-call semantics the batch's own doc
/// comment states ("isolated outputs, identical to a standalone
/// `call_ink_function`").
#[test]
fn call_ink_functions_flushes_each_calls_command_before_the_next_call_runs() {
    use crate::BrinkFlowRequest;
    use crate::test_support::{add_story_assets, make_test_app};

    #[derive(Event, Clone, Debug)]
    struct Bump;

    impl BrinkCommand for Bump {
        fn from_ink_args(_args: &[Value]) -> Result<Self, BrinkArgError> {
            Ok(Self)
        }
    }

    #[derive(Resource, Default)]
    struct Counter(i32);

    #[expect(
        clippy::needless_pass_by_value,
        reason = "bind_brink_query system param, not a plain function argument"
    )]
    fn read_counter(In((_entity, _args)): In<BrinkQueryInput>, counter: Res<Counter>) -> Value {
        Value::Int(counter.0)
    }

    let mut app = make_test_app();
    app.bind_brink_command::<(), Bump>("bump");
    app.bind_brink_query::<(), _, _>("read_counter", read_counter);
    app.init_resource::<Counter>();
    app.add_observer(|_on: On<Bump>, mut counter: ResMut<Counter>| {
        counter.0 += 1;
    });

    let (program, tables, ctx) = compile_test_story(
        "EXTERNAL bump()\n\
         EXTERNAL read_counter()\n\
         -> END\n\
         === function do_bump() ===\n\
         ~ return bump()\n\
         === function do_read() ===\n\
         ~ return read_counter()\n",
    );
    let story = add_story_assets(&mut app, program, tables, ctx);
    let entity = app
        .world_mut()
        .spawn(BrinkFlowRequest::<()>::builder().story(story).build())
        .id();
    app.update();

    let calls: Vec<(&str, Vec<Value>)> = vec![("do_bump", vec![]), ("do_read", vec![])];
    let results: Vec<Value> = call_ink_functions::<(), _, _>(app.world_mut(), entity, calls)
        .into_iter()
        .map(|r| r.unwrap())
        .collect();

    assert_eq!(
        results[1],
        Value::Int(1),
        "do_read (call 2) must observe do_bump's (call 1) World effect — proving \
             call_ink_functions flushes each call's command triggers before the next \
             call runs, not once at the end of the batch; got {results:?}"
    );
}