swactor/crates/wasm-actor/tests/wasm_actor.rs

use swactor::actor::{ActorAddress, ActorInterface};
use swactor::runtime::{Ctx, Runtime, RuntimeConfig};
use swactor_wasm_actor::{ByteMessage, SharedEngine, WasmActorBuilder, WasmActorError};

use proptest::prelude::*;

fn guest_wasm(name: &str) -> Vec<u8> {
    let path = format!(
        "{}/tests/guests/{name}/target/wasm32-unknown-unknown/release/{name}_guest.wasm",
        env!("CARGO_MANIFEST_DIR")
    );
    std::fs::read(&path).unwrap_or_else(|e| panic!("failed to read {path}: {e}"))
}

/// Build a message with an inbox address prepended (the guest contract).
fn framed_msg(dest: &ActorAddress, payload: &[u8]) -> ByteMessage {
    let mut buf = Vec::with_capacity(32 + payload.len());
    buf.extend_from_slice(&dest.0);
    buf.extend_from_slice(payload);
    ByteMessage(buf)
}

// ── Echo: send bytes in, same bytes come back ────────────────────────────────

#[test]
fn echo_returns_same_payload() {
    let engine = SharedEngine::new().unwrap();
    let actor = WasmActorBuilder::new(engine, guest_wasm("echo"))
        .build()
        .unwrap();

    let rt = Runtime::new(RuntimeConfig::default());
    let inbox = rt.new_inbox::<ByteMessage>().unwrap();
    let addr = rt.spawn(actor).unwrap();

    let payload = b"hello wasm";
    rt.send_to(addr, framed_msg(inbox.addr(), payload)).unwrap();
    rt.tick();

    let received = inbox.try_recv().expect("inbox should have a message");
    assert_eq!(received.0, payload);
}

#[test]
fn echo_preserves_binary_payload() {
    let engine = SharedEngine::new().unwrap();
    let actor = WasmActorBuilder::new(engine, guest_wasm("echo"))
        .build()
        .unwrap();

    let rt = Runtime::new(RuntimeConfig::default());
    let inbox = rt.new_inbox::<ByteMessage>().unwrap();
    let addr = rt.spawn(actor).unwrap();

    let payload: Vec<u8> = (0..=255).collect();
    rt.send_to(addr, framed_msg(inbox.addr(), &payload)).unwrap();
    rt.tick();

    let received = inbox.try_recv().expect("inbox should have a message");
    assert_eq!(received.0, payload);
}

// ── Silent: processes messages without sending anything ───────────────────────

#[test]
fn silent_produces_no_output() {
    let engine = SharedEngine::new().unwrap();
    let actor = WasmActorBuilder::new(engine, guest_wasm("silent"))
        .build()
        .unwrap();

    let rt = Runtime::new(RuntimeConfig::default());
    let inbox = rt.new_inbox::<ByteMessage>().unwrap();
    let addr = rt.spawn(actor).unwrap();

    rt.send_to(addr, ByteMessage(b"ignored".to_vec())).unwrap();
    rt.tick();

    assert!(inbox.try_recv().is_none(), "silent guest should not send anything");
}

// ── Double: one message in, two messages out ─────────────────────────────────

#[test]
fn double_sends_two_copies() {
    let engine = SharedEngine::new().unwrap();
    let actor = WasmActorBuilder::new(engine, guest_wasm("double"))
        .build()
        .unwrap();

    let rt = Runtime::new(RuntimeConfig::default());
    let inbox = rt.new_inbox::<ByteMessage>().unwrap();
    let addr = rt.spawn(actor).unwrap();

    let payload = b"dup me";
    rt.send_to(addr, framed_msg(inbox.addr(), payload)).unwrap();
    rt.tick();

    let first = inbox.try_recv().expect("should receive first copy");
    let second = inbox.try_recv().expect("should receive second copy");
    assert_eq!(first.0, payload);
    assert_eq!(second.0, payload);
    assert!(inbox.try_recv().is_none(), "exactly two messages expected");
}

// ── Missing export → WasmActorError::MissingExport ───────────────────────────

#[test]
fn missing_alloc_export_returns_error() {
    // Minimal valid Wasm module: (module) — no exports at all
    let minimal_wasm = wat::parse_str("(module)").unwrap();
    let engine = SharedEngine::new().unwrap();
    let result = WasmActorBuilder::new(engine, minimal_wasm).build();
    match result {
        Err(WasmActorError::MissingExport(name)) => {
            assert!(
                name == "memory" || name == "alloc",
                "expected missing memory or alloc, got: {name}"
            );
        }
        Err(other) => panic!("expected MissingExport, got: {other}"),
        Ok(_) => panic!("expected error for module with no exports"),
    }
}

// ── Engine sharing: two actors from the same engine ──────────────────────────

#[test]
fn shared_engine_serves_multiple_actors() {
    let engine = SharedEngine::new().unwrap();

    let echo = WasmActorBuilder::new(engine.clone(), guest_wasm("echo"))
        .build()
        .unwrap();
    let silent = WasmActorBuilder::new(engine, guest_wasm("silent"))
        .build()
        .unwrap();

    let rt = Runtime::new(RuntimeConfig::default());
    let inbox = rt.new_inbox::<ByteMessage>().unwrap();

    let echo_addr = rt.spawn(echo).unwrap();
    let _silent_addr = rt.spawn(silent).unwrap();

    let payload = b"shared engine test";
    rt.send_to(echo_addr, framed_msg(inbox.addr(), payload)).unwrap();
    rt.tick();

    let received = inbox.try_recv().expect("echo actor should still work");
    assert_eq!(received.0, payload);
}

// ── Safety: edge cases that previously caused panics or corruption ────────────

#[test]
fn oob_send_traps_cleanly_and_actor_survives() {
    // Guest calls swactor.send with dest_ptr pointing past the end of memory.
    // The host should trap the call; the actor should survive for future messages.
    let wat = r#"
        (module
            (import "swactor" "send" (func $send (param i32 i32 i32)))
            (memory (export "memory") 1)
            (func (export "alloc") (param i32) (result i32)
                i32.const 0  ;; return start of memory (simplistic)
            )
            (func (export "handle") (param i32 i32)
                ;; Call send with dest_ptr = 65536 (1 page = end of memory, OOB for 32 bytes)
                i32.const 65536
                i32.const 0
                i32.const 0
                call $send
            )
        )
    "#;
    let wasm = wat::parse_str(wat).unwrap();
    let engine = SharedEngine::new().unwrap();
    let actor = WasmActorBuilder::new(engine, wasm).build().unwrap();

    let rt = Runtime::new(RuntimeConfig::default());
    let inbox = rt.new_inbox::<ByteMessage>().unwrap();
    let addr = rt.spawn(actor).unwrap();

    // Send a message — handle will try OOB send, which traps
    rt.send_to(addr, ByteMessage(vec![42])).unwrap();
    rt.tick();

    // No message should arrive (the send was invalid)
    assert!(inbox.try_recv().is_none(), "OOB send should not produce a message");
}

#[test]
fn alloc_oom_drops_message_actor_stays_alive() {
    // Guest alloc always returns 0 (OOM). Message should be dropped,
    // actor should remain alive for subsequent messages.
    let wat = r#"
        (module
            (import "swactor" "send" (func $send (param i32 i32 i32)))
            (memory (export "memory") 1)
            (func (export "alloc") (param i32) (result i32)
                i32.const 0  ;; always OOM
            )
            (func (export "handle") (param i32 i32)
                ;; Should never be called if alloc returned 0 for non-zero len
            )
        )
    "#;
    let wasm = wat::parse_str(wat).unwrap();
    let engine = SharedEngine::new().unwrap();
    let actor = WasmActorBuilder::new(engine, wasm).build().unwrap();

    let rt = Runtime::new(RuntimeConfig::default());
    let addr = rt.spawn(actor).unwrap();

    // Send a non-empty message — alloc returns 0, message should be dropped
    rt.send_to(addr, ByteMessage(vec![1, 2, 3])).unwrap();
    rt.tick();

    // Actor is still alive — send another message, tick again (no panic)
    rt.send_to(addr, ByteMessage(vec![4, 5, 6])).unwrap();
    rt.tick();
}

#[test]
fn negative_alloc_ptr_drops_message() {
    // Guest alloc returns -1. Host should detect the negative pointer and drop.
    let wat = r#"
        (module
            (import "swactor" "send" (func $send (param i32 i32 i32)))
            (memory (export "memory") 1)
            (func (export "alloc") (param i32) (result i32)
                i32.const -1  ;; invalid negative pointer
            )
            (func (export "handle") (param i32 i32))
        )
    "#;
    let wasm = wat::parse_str(wat).unwrap();
    let engine = SharedEngine::new().unwrap();
    let actor = WasmActorBuilder::new(engine, wasm).build().unwrap();

    let rt = Runtime::new(RuntimeConfig::default());
    let addr = rt.spawn(actor).unwrap();

    rt.send_to(addr, ByteMessage(vec![1])).unwrap();
    rt.tick(); // should not panic

    // Actor survives
    rt.send_to(addr, ByteMessage(vec![2])).unwrap();
    rt.tick();
}

#[test]
fn handle_trap_drops_message_actor_survives() {
    // Guest handle executes `unreachable`, causing a Wasm trap.
    // Message should be dropped, actor should stay alive.
    let wat = r#"
        (module
            (import "swactor" "send" (func $send (param i32 i32 i32)))
            (memory (export "memory") 1)
            (func (export "alloc") (param i32) (result i32)
                i32.const 256  ;; valid allocation
            )
            (func (export "handle") (param i32 i32)
                unreachable  ;; trap!
            )
        )
    "#;
    let wasm = wat::parse_str(wat).unwrap();
    let engine = SharedEngine::new().unwrap();
    let actor = WasmActorBuilder::new(engine, wasm).build().unwrap();

    let rt = Runtime::new(RuntimeConfig::default());
    let addr = rt.spawn(actor).unwrap();

    rt.send_to(addr, ByteMessage(vec![1, 2, 3])).unwrap();
    rt.tick(); // handle traps, but actor should survive

    // Actor is still alive
    rt.send_to(addr, ByteMessage(vec![4, 5, 6])).unwrap();
    rt.tick();
}

// ── Bounds safety: alloc pointer near end of linear memory ────────────────────

#[test]
fn alloc_near_end_of_memory_drops_message_actor_survives() {
    // Guest alloc returns 65500 (near end of 1-page / 65536-byte memory).
    // A 100-byte message means ptr+len = 65600, which exceeds memory bounds.
    // The actor should drop the message and survive — same as any other
    // allocation failure — rather than being permanently killed.
    let wat = r#"
        (module
            (import "swactor" "send" (func $send (param i32 i32 i32)))
            (memory (export "memory") 1)
            (func (export "alloc") (param i32) (result i32)
                i32.const 65500  ;; near end of 64KiB memory
            )
            (func (export "handle") (param i32 i32)
                ;; should never be reached if bounds check works
            )
        )
    "#;
    let wasm = wat::parse_str(wat).unwrap();
    let engine = SharedEngine::new().unwrap();
    let actor = WasmActorBuilder::new(engine, wasm).build().unwrap();

    let rt = Runtime::new(RuntimeConfig::default());
    let addr = rt.spawn(actor).unwrap();

    // Send a message whose length exceeds the remaining space at ptr 65500
    rt.send_to(addr, ByteMessage(vec![0u8; 100])).unwrap();
    rt.tick();

    // The actor should still be alive — send another message and tick without panic
    rt.send_to(addr, ByteMessage(vec![1u8; 10])).unwrap();
    rt.tick();
}

// ── Edge cases: empty and oversized messages ─────────────────────────────────

#[test]
fn empty_message_is_handled_without_crash() {
    // A zero-length ByteMessage should pass through the alloc/handle pipeline
    // without crashing. The echo guest returns nothing (len < 32 guard).
    let engine = SharedEngine::new().unwrap();
    let actor = WasmActorBuilder::new(engine, guest_wasm("echo"))
        .build()
        .unwrap();

    let rt = Runtime::new(RuntimeConfig::default());
    let inbox = rt.new_inbox::<ByteMessage>().unwrap();
    let addr = rt.spawn(actor).unwrap();

    rt.send_to(addr, ByteMessage(vec![])).unwrap();
    rt.tick();

    // Echo guest does `if len < 32 { return; }` — so no reply expected
    assert!(inbox.try_recv().is_none(), "empty message should produce no reply");

    // Actor survives — can still process a real message
    let payload = b"still alive";
    rt.send_to(addr, framed_msg(inbox.addr(), payload)).unwrap();
    rt.tick();
    let received = inbox.try_recv().expect("actor should still be alive");
    assert_eq!(received.0, payload);
}

#[test]
fn message_larger_than_linear_memory_is_dropped() {
    // A message of 65537 bytes exceeds the 1-page (64KiB) guest memory.
    // Guest alloc will OOM (return 0) → message dropped, actor survives.
    let engine = SharedEngine::new().unwrap();
    let actor = WasmActorBuilder::new(engine, guest_wasm("echo"))
        .build()
        .unwrap();

    let rt = Runtime::new(RuntimeConfig::default());
    let inbox = rt.new_inbox::<ByteMessage>().unwrap();
    let addr = rt.spawn(actor).unwrap();

    rt.send_to(addr, ByteMessage(vec![0u8; 65537])).unwrap();
    rt.tick();

    assert!(inbox.try_recv().is_none(), "oversized message should be dropped");

    // Actor survives
    let payload = b"after oversize";
    rt.send_to(addr, framed_msg(inbox.addr(), payload)).unwrap();
    rt.tick();
    let received = inbox.try_recv().expect("actor should survive oversized message");
    assert_eq!(received.0, payload);
}

// ── Sustained load: bump allocator exhaustion ────────────────────────────────

#[test]
fn sequential_messages_degrade_gracefully_after_allocator_exhaustion() {
    // The echo guest has a 64KiB bump allocator that never frees. Under
    // sustained load, alloc eventually returns 0 (OOM) and messages are
    // silently dropped. The actor must survive throughout — no panics,
    // no poisoning.
    let engine = SharedEngine::new().unwrap();
    let actor = WasmActorBuilder::new(engine, guest_wasm("echo"))
        .build()
        .unwrap();

    let rt = Runtime::new(RuntimeConfig::default());
    let inbox = rt.new_inbox::<ByteMessage>().unwrap();
    let addr = rt.spawn(actor).unwrap();

    let payload = b"ping";
    let mut echoed = 0usize;

    // Send enough messages to exhaust the 64KiB heap.
    // Each framed message is 36 bytes (32 addr + 4 payload), aligned to 40.
    // 65536 / 40 = ~1638, but heap offset within memory varies. Send 2000.
    for _ in 0..2000 {
        rt.send_to(addr, framed_msg(inbox.addr(), payload)).unwrap();
        rt.tick();
        if inbox.try_recv().is_some() {
            echoed += 1;
        }
    }

    // Some messages were echoed before OOM
    assert!(echoed > 0, "should echo at least some messages");
    // After OOM, messages were dropped — so not all 2000 echoed
    assert!(echoed < 2000, "allocator should exhaust before 2000 messages");
}

// ── Fire-and-forget: guest sends to nonexistent address ──────────────────────

#[test]
fn guest_send_to_nonexistent_address_is_silently_dropped() {
    // Guest sends to an all-zero 32-byte address that isn't registered in
    // the runtime. The ctx.send() error is silently dropped (fire-and-forget).
    // Actor must survive.
    let wat = r#"
        (module
            (import "swactor" "send" (func $send (param i32 i32 i32)))
            (memory (export "memory") 1)
            (func (export "alloc") (param i32) (result i32)
                i32.const 0
            )
            (func (export "handle") (param i32 i32)
                ;; send to address at offset 0 (all zeros — no such actor)
                ;; with 1-byte payload at offset 32
                i32.const 0    ;; dest_ptr
                i32.const 32   ;; payload_ptr
                i32.const 1    ;; payload_len
                call $send
            )
        )
    "#;
    let wasm = wat::parse_str(wat).unwrap();
    let engine = SharedEngine::new().unwrap();
    let actor = WasmActorBuilder::new(engine, wasm).build().unwrap();

    let rt = Runtime::new(RuntimeConfig::default());
    let addr = rt.spawn(actor).unwrap();

    rt.send_to(addr, ByteMessage(vec![42])).unwrap();
    rt.tick(); // guest sends to nonexistent address — should not panic

    // Actor survives
    rt.send_to(addr, ByteMessage(vec![99])).unwrap();
    rt.tick();
}

// ── Builder validation: wrong export signatures ──────────────────────────────

#[test]
fn wrong_handle_signature_is_rejected() {
    // Module exports `handle` with wrong signature: (i32) -> i32 instead of (i32, i32) -> ()
    // Builder maps get_typed_func errors to MissingExport (signature mismatch = not found).
    let wat = r#"
        (module
            (memory (export "memory") 1)
            (func (export "alloc") (param i32) (result i32) i32.const 0)
            (func (export "handle") (param i32) (result i32) i32.const 0)
        )
    "#;
    let wasm = wat::parse_str(wat).unwrap();
    let engine = SharedEngine::new().unwrap();
    let result = WasmActorBuilder::new(engine, wasm).build();
    assert!(result.is_err(), "should reject wrong handle signature");
}

#[test]
fn wrong_memory_export_name_returns_missing_export() {
    // Module has a memory, but exported as "mem" instead of "memory"
    let wat = r#"
        (module
            (memory (export "mem") 1)
            (func (export "alloc") (param i32) (result i32) i32.const 0)
            (func (export "handle") (param i32 i32))
        )
    "#;
    let wasm = wat::parse_str(wat).unwrap();
    let engine = SharedEngine::new().unwrap();
    let result = WasmActorBuilder::new(engine, wasm).build();
    match result {
        Err(WasmActorError::MissingExport("memory")) => {} // expected
        Err(other) => panic!("expected MissingExport(\"memory\"), got: {other}"),
        Ok(_) => panic!("should reject module without 'memory' export"),
    }
}

// ── Lifecycle: graceful stop of WASM actor ───────────────────────────────────

#[test]
fn graceful_stop_cleans_up_wasm_actor() {
    // After stopping a WASM actor, it should be removed from the runtime.
    // The wasmtime Store is dropped cleanly (no leak, no crash).
    let engine = SharedEngine::new().unwrap();
    let actor = WasmActorBuilder::new(engine, guest_wasm("echo"))
        .build()
        .unwrap();

    let rt = Runtime::new(RuntimeConfig::default());
    let inbox = rt.new_inbox::<ByteMessage>().unwrap();
    let addr = rt.spawn(actor).unwrap();

    // Verify the actor works
    let payload = b"before stop";
    rt.send_to(addr, framed_msg(inbox.addr(), payload)).unwrap();
    rt.tick();
    let received = inbox.try_recv().expect("actor should echo before stop");
    assert_eq!(received.0, payload);

    // Stop the actor
    rt.stop_actor(addr).unwrap();
    rt.tick(); // process stop signal
    rt.tick(); // cleanup_dead phase

    // Actor is gone — send should fail
    let result = rt.send_to(addr, ByteMessage(vec![1]));
    assert!(result.is_err(), "send to stopped actor should fail");
}

// ── Host import validation: negative payload_len ─────────────────────────────

#[test]
fn negative_payload_len_in_send_traps_actor_survives() {
    // Guest calls swactor.send with payload_len = -1. The host import
    // should trap (negative argument check), and the actor should survive.
    let wat = r#"
        (module
            (import "swactor" "send" (func $send (param i32 i32 i32)))
            (memory (export "memory") 1)
            (func (export "alloc") (param i32) (result i32)
                i32.const 256
            )
            (func (export "handle") (param i32 i32)
                i32.const 0    ;; dest_ptr
                i32.const 0    ;; payload_ptr
                i32.const -1   ;; payload_len (negative!)
                call $send
            )
        )
    "#;
    let wasm = wat::parse_str(wat).unwrap();
    let engine = SharedEngine::new().unwrap();
    let actor = WasmActorBuilder::new(engine, wasm).build().unwrap();

    let rt = Runtime::new(RuntimeConfig::default());
    let addr = rt.spawn(actor).unwrap();

    rt.send_to(addr, ByteMessage(vec![1, 2, 3])).unwrap();
    rt.tick(); // guest calls send with negative len — should trap

    // Actor survives
    rt.send_to(addr, ByteMessage(vec![4, 5, 6])).unwrap();
    rt.tick();
}

// ── Independent stores: two echo actors from same engine + bytes ─────────────

#[test]
fn two_echo_actors_from_same_engine_are_independent() {
    let engine = SharedEngine::new().unwrap();
    let wasm_bytes = guest_wasm("echo");

    let actor_a = WasmActorBuilder::new(engine.clone(), wasm_bytes.clone())
        .build()
        .unwrap();
    let actor_b = WasmActorBuilder::new(engine, wasm_bytes)
        .build()
        .unwrap();

    let rt = Runtime::new(RuntimeConfig::default());
    let inbox_a = rt.new_inbox::<ByteMessage>().unwrap();
    let inbox_b = rt.new_inbox::<ByteMessage>().unwrap();

    let addr_a = rt.spawn(actor_a).unwrap();
    let addr_b = rt.spawn(actor_b).unwrap();

    // Send different payloads to each
    rt.send_to(addr_a, framed_msg(inbox_a.addr(), b"for-a")).unwrap();
    rt.send_to(addr_b, framed_msg(inbox_b.addr(), b"for-b")).unwrap();
    rt.tick();

    let recv_a = inbox_a.try_recv().expect("actor A should echo");
    let recv_b = inbox_b.try_recv().expect("actor B should echo");
    assert_eq!(recv_a.0, b"for-a");
    assert_eq!(recv_b.0, b"for-b");

    // Cross-check: no leakage between actors
    assert!(inbox_a.try_recv().is_none());
    assert!(inbox_b.try_recv().is_none());
}

// ── Watch integration: native watcher observes WASM actor death ──────────────

struct ExitWatcher {
    exit_count: std::sync::Arc<std::sync::atomic::AtomicUsize>,
    last_reason: std::sync::Arc<std::sync::Mutex<Option<swactor::actor::ExitReason>>>,
}

#[derive(Clone)]
struct WatchThis(ActorAddress);

impl ActorInterface for ExitWatcher {
    type Incoming = WatchThis;
    type Response = ();
    fn handle(&mut self, ctx: &Ctx, msg: WatchThis) {
        ctx.watch(msg.0);
    }
    fn on_actor_exit(&mut self, _ctx: &Ctx, exited: swactor::actor::ActorExited) {
        self.exit_count.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
        *self.last_reason.lock().unwrap() = Some(exited.reason);
    }
}

#[test]
fn native_watcher_notified_when_wasm_actor_stops() {
    // A native actor watches a WASM actor. When the WASM actor is stopped,
    // the watcher should receive ActorExited with ExitReason::Stopped.
    let engine = SharedEngine::new().unwrap();
    let wasm = WasmActorBuilder::new(engine, guest_wasm("echo"))
        .build()
        .unwrap();

    let rt = Runtime::new(RuntimeConfig::default());

    let exit_count = std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0));
    let last_reason = std::sync::Arc::new(std::sync::Mutex::new(None));
    let watcher = ExitWatcher {
        exit_count: exit_count.clone(),
        last_reason: last_reason.clone(),
    };

    let wasm_addr = rt.spawn(wasm).unwrap();
    let watcher_addr = rt.spawn(watcher).unwrap();

    // Tell watcher to watch the WASM actor
    rt.send_to(watcher_addr, WatchThis(wasm_addr)).unwrap();
    for _ in 0..3 { rt.tick(); }

    // Stop the WASM actor
    rt.stop_actor(wasm_addr).unwrap();
    for _ in 0..5 { rt.tick(); }

    assert_eq!(exit_count.load(std::sync::atomic::Ordering::SeqCst), 1);
    assert_eq!(
        *last_reason.lock().unwrap(),
        Some(swactor::actor::ExitReason::Stopped)
    );
}

// ── WASM-to-WASM: two WASM actors communicating ─────────────────────────────

#[test]
fn wasm_to_wasm_message_relay() {
    // Echo A echoes to Echo B's address, Echo B echoes to an external inbox.
    // This verifies the full WASM→runtime→WASM→runtime→inbox path.
    let engine = SharedEngine::new().unwrap();
    let wasm_bytes = guest_wasm("echo");

    let actor_a = WasmActorBuilder::new(engine.clone(), wasm_bytes.clone())
        .build()
        .unwrap();
    let actor_b = WasmActorBuilder::new(engine, wasm_bytes)
        .build()
        .unwrap();

    let rt = Runtime::new(RuntimeConfig::default());
    let inbox = rt.new_inbox::<ByteMessage>().unwrap();

    let addr_a = rt.spawn(actor_a).unwrap();
    let addr_b = rt.spawn(actor_b).unwrap();

    // Send to actor A: "echo your payload to actor B"
    // Actor A receives [addr_b | payload_for_b]
    // Actor A echoes payload_for_b to addr_b
    // payload_for_b itself is [inbox_addr | final_payload]
    // Actor B receives [inbox_addr | final_payload]
    // Actor B echoes final_payload to inbox
    let final_payload = b"relayed";
    let payload_for_b = framed_msg(inbox.addr(), final_payload);
    let msg_for_a = framed_msg(&addr_b, &payload_for_b.0);

    rt.send_to(addr_a, msg_for_a).unwrap();
    rt.tick(); // A receives, echoes to B
    rt.tick(); // B receives, echoes to inbox

    let received = inbox.try_recv().expect("should receive relayed message");
    assert_eq!(received.0, final_payload);
}

// ── Integration: WasmActor alongside a native Rust actor ─────────────────────

#[derive(Clone)]
struct ForwardToWasm {
    wasm_addr: ActorAddress,
    inbox_addr: ActorAddress,
}

struct Forwarder;

impl ActorInterface for Forwarder {
    type Incoming = ForwardToWasm;
    type Response = ();

    fn handle(&mut self, ctx: &Ctx, msg: ForwardToWasm) {
        // Build the framed message and forward to the wasm actor
        let payload = b"from native";
        let framed = framed_msg(&msg.inbox_addr, payload);
        let _ = ctx.send(msg.wasm_addr, framed);
    }
}

#[test]
fn native_actor_communicates_with_wasm_actor() {
    let engine = SharedEngine::new().unwrap();
    let wasm = WasmActorBuilder::new(engine, guest_wasm("echo"))
        .build()
        .unwrap();

    let rt = Runtime::new(RuntimeConfig::default());
    let inbox = rt.new_inbox::<ByteMessage>().unwrap();

    let wasm_addr = rt.spawn(wasm).unwrap();
    let forwarder_addr = rt.spawn(Forwarder).unwrap();

    rt.send_to(
        forwarder_addr,
        ForwardToWasm {
            wasm_addr,
            inbox_addr: *inbox.addr(),
        },
    )
    .unwrap();

    // Tick 1: Forwarder receives message and sends to WasmActor
    rt.tick();
    // Tick 2: WasmActor receives the forwarded message and echoes to inbox
    rt.tick();

    let received = inbox.try_recv().expect("wasm actor should have echoed");
    assert_eq!(received.0, b"from native");
}

// ── Stale outbox: sends before trap leak into next handle ─────────────────────

#[test]
fn outbox_entries_from_trapped_handle_do_not_leak_into_next_call() {
    // A guest that calls swactor.send() successfully, then traps.
    // The outbox contains the send from before the trap.
    // On the next handle call (which succeeds without sending), the stale
    // outbox entry should NOT be delivered.
    //
    // Counter incremented BEFORE the if-branch so it persists past the trap.
    let wat = r#"
        (module
            (import "swactor" "send" (func $send (param i32 i32 i32)))
            (memory (export "memory") 1)
            (global $counter (mut i32) (i32.const 0))

            (func (export "alloc") (param i32) (result i32)
                i32.const 256
            )
            (func (export "handle") (param $ptr i32) (param $len i32)
                ;; Increment counter first (survives trap)
                global.get $counter
                i32.const 1
                i32.add
                global.set $counter

                ;; If counter was 0 (now 1): send then trap
                global.get $counter
                i32.const 1
                i32.eq
                if
                    local.get $ptr  ;; dest_ptr (first 32 bytes = inbox address)
                    i32.const 32    ;; payload_ptr
                    i32.const 1     ;; payload_len
                    call $send
                    unreachable     ;; trap after send
                end
                ;; counter > 1: do nothing (no send)
            )
        )
    "#;
    let wasm = wat::parse_str(wat).unwrap();
    let engine = SharedEngine::new().unwrap();
    let actor = WasmActorBuilder::new(engine, wasm).build().unwrap();

    let rt = Runtime::new(RuntimeConfig::default());
    let inbox = rt.new_inbox::<ByteMessage>().unwrap();
    let addr = rt.spawn(actor).unwrap();

    // First message: guest sends to outbox then traps — stale entry in outbox
    // Use framed_msg so the first 32 bytes are the inbox address
    rt.send_to(addr, framed_msg(inbox.addr(), b"x")).unwrap();
    rt.tick();

    // No message should have been delivered (handle trapped before outbox drain)
    assert!(inbox.try_recv().is_none(), "trapped handle should not deliver messages");

    // Second message: guest does nothing (counter=2, no send, no trap).
    // If the outbox wasn't cleared, the stale entry would be drained here.
    rt.send_to(addr, framed_msg(inbox.addr(), b"y")).unwrap();
    rt.tick();

    // Should still be empty — the stale outbox entry must not leak
    assert!(
        inbox.try_recv().is_none(),
        "stale outbox entry from trapped call should not leak into next handle"
    );
}

// ── Builder validation: invalid WASM bytes ───────────────────────────────────

#[test]
fn invalid_wasm_bytes_returns_wasmtime_error() {
    let garbage = vec![0u8, 1, 2, 3]; // not valid wasm
    let engine = SharedEngine::new().unwrap();
    let result = WasmActorBuilder::new(engine, garbage).build();
    match result {
        Err(WasmActorError::Wasmtime(_)) => {} // expected — compilation failure
        Err(other) => panic!("expected Wasmtime error for invalid bytes, got: {other}"),
        Ok(_) => panic!("should reject invalid wasm bytes"),
    }
}

// ── Guest sends zero-length payload ──────────────────────────────────────────

#[test]
fn guest_send_with_zero_length_payload_delivers_empty_message() {
    // Guest calls swactor.send with payload_len=0. This should produce
    // a ByteMessage(vec![]) at the destination.
    let wat = r#"
        (module
            (import "swactor" "send" (func $send (param i32 i32 i32)))
            (memory (export "memory") 1)
            (func (export "alloc") (param i32) (result i32)
                i32.const 256  ;; valid allocation
            )
            (func (export "handle") (param $ptr i32) (param $len i32)
                ;; Send with the first 32 bytes as dest, zero-length payload
                local.get $ptr
                i32.const 32   ;; payload_ptr (doesn't matter, len is 0)
                i32.const 0    ;; payload_len
                call $send
            )
        )
    "#;
    let wasm = wat::parse_str(wat).unwrap();
    let engine = SharedEngine::new().unwrap();
    let actor = WasmActorBuilder::new(engine, wasm).build().unwrap();

    let rt = Runtime::new(RuntimeConfig::default());
    let inbox = rt.new_inbox::<ByteMessage>().unwrap();
    let addr = rt.spawn(actor).unwrap();

    // Build a framed message with the inbox address as the first 32 bytes
    let msg = framed_msg(inbox.addr(), b"ignored-payload");
    rt.send_to(addr, msg).unwrap();
    rt.tick();

    let received = inbox.try_recv().expect("should receive zero-length message");
    assert!(received.0.is_empty(), "payload should be empty");
}

// ── Multiple sequential traps: actor survives repeated failures ──────────────

#[test]
fn actor_survives_multiple_sequential_traps() {
    // After 3 consecutive traps, the actor should still be alive and
    // able to process a non-trapping message.
    let wat = r#"
        (module
            (import "swactor" "send" (func $send (param i32 i32 i32)))
            (memory (export "memory") 1)
            (global $counter (mut i32) (i32.const 0))
            (func (export "alloc") (param i32) (result i32)
                i32.const 256
            )
            (func (export "handle") (param $ptr i32) (param $len i32)
                global.get $counter
                i32.const 3
                i32.lt_u
                if
                    ;; First 3 calls: trap
                    global.get $counter
                    i32.const 1
                    i32.add
                    global.set $counter
                    unreachable
                end
                ;; 4th+ call: echo the message back using first 32 bytes as dest
                local.get $ptr
                local.get $ptr
                i32.const 32
                i32.add
                local.get $len
                i32.const 32
                i32.sub
                call $send
            )
        )
    "#;
    let wasm = wat::parse_str(wat).unwrap();
    let engine = SharedEngine::new().unwrap();
    let actor = WasmActorBuilder::new(engine, wasm).build().unwrap();

    let rt = Runtime::new(RuntimeConfig::default());
    let inbox = rt.new_inbox::<ByteMessage>().unwrap();
    let addr = rt.spawn(actor).unwrap();

    // 3 trapping messages
    for _ in 0..3 {
        rt.send_to(addr, framed_msg(inbox.addr(), b"will trap")).unwrap();
        rt.tick();
        assert!(inbox.try_recv().is_none(), "trapped call should produce nothing");
    }

    // 4th message: should succeed
    let payload = b"survived";
    rt.send_to(addr, framed_msg(inbox.addr(), payload)).unwrap();
    rt.tick();

    let received = inbox.try_recv().expect("actor should work after multiple traps");
    assert_eq!(received.0, payload);
}

// ── Builder: module with start function that traps ───────────────────────────

#[test]
fn module_with_trapping_start_function_returns_error() {
    // WASM modules can have a (start) function that runs during instantiation.
    // If it traps, build() should return an error.
    let wat = r#"
        (module
            (memory (export "memory") 1)
            (func (export "alloc") (param i32) (result i32) i32.const 256)
            (func (export "handle") (param i32 i32))
            (func $init unreachable)
            (start $init)
        )
    "#;
    let wasm = wat::parse_str(wat).unwrap();
    let engine = SharedEngine::new().unwrap();
    let result = WasmActorBuilder::new(engine, wasm).build();
    assert!(result.is_err(), "module with trapping start function should fail to build");
}

// ── Self-send: guest sends message back to own address ───────────────────────

#[test]
fn guest_self_send_creates_feedback_loop() {
    // Echo guest sends its payload to a destination. If we set the dest
    // to the actor's OWN address, it creates a feedback loop. The actor
    // should process the self-sent message on the next tick.
    let engine = SharedEngine::new().unwrap();
    let actor = WasmActorBuilder::new(engine, guest_wasm("echo"))
        .build()
        .unwrap();

    let rt = Runtime::new(RuntimeConfig::default());
    let inbox = rt.new_inbox::<ByteMessage>().unwrap();
    let addr = rt.spawn(actor).unwrap();

    // Frame: dest=self, payload=[inbox_addr | "hello"]
    // Echo will send [inbox_addr | "hello"] back to itself.
    // On next tick, it receives [inbox_addr | "hello"], echoes "hello" to inbox.
    let inner_msg = framed_msg(inbox.addr(), b"hello");
    let self_msg = framed_msg(&addr, &inner_msg.0);

    rt.send_to(addr, self_msg).unwrap();
    rt.tick(); // actor echoes inner_msg to self
    rt.tick(); // actor receives inner_msg, echoes "hello" to inbox

    let received = inbox.try_recv().expect("should receive after self-send loop");
    assert_eq!(received.0, b"hello");
}

// ── Amplification: guest sends many messages in one handle ───────────────────

#[test]
fn guest_sending_many_messages_in_one_handle_all_delivered() {
    // A guest that calls swactor.send N times in a single handle call.
    // All N messages should be delivered via the outbox drain.
    let wat = r#"
        (module
            (import "swactor" "send" (func $send (param i32 i32 i32)))
            (memory (export "memory") 1)
            (func (export "alloc") (param i32) (result i32)
                i32.const 256
            )
            (func (export "handle") (param $ptr i32) (param $len i32)
                ;; Send 10 messages, each with 0-byte payload
                ;; dest_ptr = $ptr (first 32 bytes of the incoming message)
                (local $i i32)
                (local.set $i (i32.const 0))
                (block $break
                    (loop $loop
                        (br_if $break (i32.ge_u (local.get $i) (i32.const 10)))
                        (call $send (local.get $ptr) (i32.const 32) (i32.const 0))
                        (local.set $i (i32.add (local.get $i) (i32.const 1)))
                        (br $loop)
                    )
                )
            )
        )
    "#;
    let wasm = wat::parse_str(wat).unwrap();
    let engine = SharedEngine::new().unwrap();
    let actor = WasmActorBuilder::new(engine, wasm).build().unwrap();

    let rt = Runtime::new(RuntimeConfig::default());
    let inbox = rt.new_inbox::<ByteMessage>().unwrap();
    let addr = rt.spawn(actor).unwrap();

    rt.send_to(addr, framed_msg(inbox.addr(), b"x")).unwrap();
    rt.tick();

    let mut count = 0;
    while inbox.try_recv().is_some() {
        count += 1;
    }
    assert_eq!(count, 10, "guest should have sent exactly 10 messages");
}

// ── Overlapping send regions: dest_ptr and payload_ptr overlap ───────────────

#[test]
fn overlapping_dest_and_payload_in_send_works() {
    // Guest calls send with dest_ptr=0, payload_ptr=16, payload_len=32.
    // The dest region [0..32] and payload region [16..48] overlap.
    // Both are read-only in the host, so this should work without corruption.
    let wat = r#"
        (module
            (import "swactor" "send" (func $send (param i32 i32 i32)))
            (memory (export "memory") 1)
            (func (export "alloc") (param i32) (result i32)
                i32.const 256
            )
            (func (export "handle") (param $ptr i32) (param $len i32)
                ;; Overlapping regions
                local.get $ptr       ;; dest_ptr (first 32 bytes of message)
                local.get $ptr
                i32.const 16
                i32.add              ;; payload_ptr = ptr + 16 (overlaps with dest)
                i32.const 32         ;; payload_len = 32
                call $send
            )
        )
    "#;
    let wasm = wat::parse_str(wat).unwrap();
    let engine = SharedEngine::new().unwrap();
    let actor = WasmActorBuilder::new(engine, wasm).build().unwrap();

    let rt = Runtime::new(RuntimeConfig::default());
    let inbox = rt.new_inbox::<ByteMessage>().unwrap();
    let addr = rt.spawn(actor).unwrap();

    // Build a message where bytes [0..32] = inbox addr, [32..] = payload
    // Guest reads dest from [0..32] (inbox addr) and payload from [16..48]
    rt.send_to(addr, framed_msg(inbox.addr(), b"overlap-test-padding!")).unwrap();
    rt.tick();

    let received = inbox.try_recv().expect("overlapping send should deliver");
    assert_eq!(received.0.len(), 32, "payload should be 32 bytes from overlapping region");
}

// ── Type mismatch: non-ByteMessage sent to WASM actor ────────────────────────

#[test]
fn non_byte_message_to_wasm_actor_is_silently_ignored() {
    // Sending a message of the wrong type (not ByteMessage) to a WASM actor.
    // The runtime's handle_any downcast fails, counting a type mismatch.
    // The actor should survive and still process valid ByteMessages.
    let engine = SharedEngine::new().unwrap();
    let actor = WasmActorBuilder::new(engine, guest_wasm("echo"))
        .build()
        .unwrap();

    let rt = Runtime::new(RuntimeConfig::default());
    let inbox = rt.new_inbox::<ByteMessage>().unwrap();
    let addr = rt.spawn(actor).unwrap();

    // Send wrong type — u32 instead of ByteMessage
    // This goes through send_any with Box::new(42u32), downcast to ByteMessage fails.
    rt.send_to(addr, 42u32).unwrap();
    rt.tick(); // type mismatch — silently ignored

    // Actor still alive — send a valid message
    let payload = b"after mismatch";
    rt.send_to(addr, framed_msg(inbox.addr(), payload)).unwrap();
    rt.tick();

    let received = inbox.try_recv().expect("actor should work after type mismatch");
    assert_eq!(received.0, payload);
}

// ── Guest state persistence: mutable global survives across messages ─────────

#[test]
fn guest_mutable_state_persists_across_messages() {
    // A guest module with a mutable global counter. Each handle call increments
    // the counter and includes it in the reply payload. Verifies that the
    // wasmtime Store and linear memory persist between handle() calls.
    let wat = r#"
        (module
            (import "swactor" "send" (func $send (param i32 i32 i32)))
            (memory (export "memory") 1)
            (global $counter (mut i32) (i32.const 0))

            (func (export "alloc") (param i32) (result i32)
                i32.const 256
            )
            (func (export "handle") (param $ptr i32) (param $len i32)
                ;; Increment counter
                global.get $counter
                i32.const 1
                i32.add
                global.set $counter

                ;; Write counter value to memory at offset 200
                (i32.store8 (i32.const 200) (global.get $counter))

                ;; Send counter byte as payload to dest at $ptr
                local.get $ptr   ;; dest_ptr (first 32 bytes of message)
                i32.const 200    ;; payload_ptr (counter byte)
                i32.const 1      ;; payload_len
                call $send
            )
        )
    "#;
    let wasm = wat::parse_str(wat).unwrap();
    let engine = SharedEngine::new().unwrap();
    let actor = WasmActorBuilder::new(engine, wasm).build().unwrap();

    let rt = Runtime::new(RuntimeConfig::default());
    let inbox = rt.new_inbox::<ByteMessage>().unwrap();
    let addr = rt.spawn(actor).unwrap();

    // Send 3 messages, each should get an incrementing counter
    for expected in 1..=3u8 {
        rt.send_to(addr, framed_msg(inbox.addr(), b"x")).unwrap();
        rt.tick();
        let received = inbox.try_recv().expect("should receive counter reply");
        assert_eq!(received.0, vec![expected], "counter should increment per message");
    }
}

// ── Spawn WASM from handler: native actor spawns WASM actor during handle ────

struct WasmSpawner {
    engine: SharedEngine,
    wasm_bytes: Vec<u8>,
}

#[derive(Clone)]
struct SpawnAndForward {
    inbox_addr: ActorAddress,
    payload: Vec<u8>,
}

impl ActorInterface for WasmSpawner {
    type Incoming = SpawnAndForward;
    type Response = ();
    fn handle(&mut self, ctx: &Ctx, msg: SpawnAndForward) {
        let actor = WasmActorBuilder::new(self.engine.clone(), self.wasm_bytes.clone())
            .build()
            .unwrap();
        let wasm_addr = ctx.spawn(actor);
        let _ = ctx.send(wasm_addr.unwrap(), framed_msg(&msg.inbox_addr, &msg.payload));
    }
}

#[test]
fn native_handler_spawns_wasm_actor_and_forwards_message() {
    let engine = SharedEngine::new().unwrap();
    let wasm_bytes = guest_wasm("echo");

    let spawner = WasmSpawner {
        engine: engine.clone(),
        wasm_bytes: wasm_bytes.clone(),
    };

    let rt = Runtime::new(RuntimeConfig::default());
    let inbox = rt.new_inbox::<ByteMessage>().unwrap();
    let spawner_addr = rt.spawn(spawner).unwrap();

    rt.send_to(
        spawner_addr,
        SpawnAndForward {
            inbox_addr: *inbox.addr(),
            payload: b"spawned-echo".to_vec(),
        },
    )
    .unwrap();

    // Tick 1: Spawner receives message, spawns WASM actor, sends to it
    rt.tick();
    // Tick 2: WASM actor processes message and echoes to inbox
    rt.tick();

    let received = inbox.try_recv().expect("dynamically spawned WASM actor should echo");
    assert_eq!(received.0, b"spawned-echo");
}

// ── Bounded mailbox: WASM actor with backpressure ────────────────────────────

#[test]
fn bounded_mailbox_applies_to_wasm_actor() {
    // With a bounded mailbox of capacity 3, sending 10 messages should
    // result in only 3 being processed (DropNewest policy).
    use swactor::runtime::MailboxOverflow;

    let engine = SharedEngine::new().unwrap();
    let actor = WasmActorBuilder::new(engine, guest_wasm("echo"))
        .build()
        .unwrap();

    let config = RuntimeConfig {
        default_mailbox_capacity: 3,
        mailbox_overflow: MailboxOverflow::DropNewest,
        ..RuntimeConfig::default()
    };
    let rt = Runtime::new(config);
    let inbox = rt.new_inbox::<ByteMessage>().unwrap();
    let addr = rt.spawn(actor).unwrap();

    // Send 10 messages before any tick — only first 3 should be kept
    for i in 0u8..10 {
        rt.send_to(addr, framed_msg(inbox.addr(), &[i])).unwrap();
    }
    rt.tick();

    let mut received = Vec::new();
    while let Some(msg) = inbox.try_recv() {
        received.push(msg.0[0]);
    }

    assert_eq!(received.len(), 3, "bounded mailbox should limit to 3 messages");
    // DropNewest keeps the first 3 sent
    assert_eq!(received, vec![0, 1, 2]);
}

// ── Property: alloc failures never kill the actor ────────────────────────────

proptest! {
    #[test]
    fn prop_any_alloc_return_value_never_kills_actor(alloc_val in -100i32..70000) {
        // Regardless of what alloc returns (negative, zero, OOB, valid),
        // sending a message should never kill the actor.
        let alloc_const = format!("i32.const {alloc_val}");
        let wat = format!(r#"
            (module
                (import "swactor" "send" (func $send (param i32 i32 i32)))
                (memory (export "memory") 1)
                (func (export "alloc") (param i32) (result i32)
                    {alloc_const}
                )
                (func (export "handle") (param i32 i32))
            )
        "#);
        let wasm = wat::parse_str(&wat).unwrap();
        let engine = SharedEngine::new().unwrap();
        let actor = WasmActorBuilder::new(engine, wasm).build().unwrap();

        let rt = Runtime::new(RuntimeConfig::default());
        let addr = rt.spawn(actor).unwrap();

        // Send a message — should never panic or poison
        rt.send_to(addr, ByteMessage(vec![0u8; 100])).unwrap();
        rt.tick();

        // Actor should still accept messages (not poisoned)
        let result = rt.send_to(addr, ByteMessage(vec![1]));
        prop_assert!(result.is_ok(), "actor should survive any alloc return value: {alloc_val}");
    }
}

// ── Multi-worker: WASM actors across threads ─────────────────────────────────

#[test]
fn wasm_actor_works_on_multi_worker_runtime() {
    // Spawn a WASM echo actor on a 2-worker runtime and verify message
    // round-trip works across threads. This is a smoke test for Send safety
    // of wasmtime Store<HostState>.
    let engine = SharedEngine::new().unwrap();
    let actor = WasmActorBuilder::new(engine, guest_wasm("echo"))
        .build()
        .unwrap();

    let config = RuntimeConfig {
        num_threads: 2,
        ..RuntimeConfig::default()
    };
    let rt = Runtime::new(config);
    let inbox = rt.new_inbox::<ByteMessage>().unwrap();
    let addr = rt.spawn(actor).unwrap();

    let payload = b"multi-worker";
    rt.send_to(addr, framed_msg(inbox.addr(), payload)).unwrap();

    // Use run() to drive the runtime on background threads
    let handle = rt.run().unwrap();
    std::thread::sleep(std::time::Duration::from_millis(50));

    let received = inbox.try_recv().expect("wasm actor should echo on MT runtime");
    assert_eq!(received.0, payload);

    handle.shutdown();
}

// ── Property-based: arbitrary bytes round-trip through echo ──────────────────

proptest! {
    #[test]
    fn prop_echo_roundtrips_arbitrary_bytes(payload in proptest::collection::vec(any::<u8>(), 0..500)) {
        let engine = SharedEngine::new().unwrap();
        let actor = WasmActorBuilder::new(engine, guest_wasm("echo"))
            .build()
            .unwrap();

        let rt = Runtime::new(RuntimeConfig::default());
        let inbox = rt.new_inbox::<ByteMessage>().unwrap();
        let addr = rt.spawn(actor).unwrap();

        let msg = framed_msg(inbox.addr(), &payload);
        rt.send_to(addr, msg).unwrap();
        rt.tick();

        if payload.is_empty() {
            // Echo guest: if total len < 32, no reply (32B addr + 0B payload = 32, but
            // the framed message is 32 + 0 = 32 bytes, and echo checks `len < 32`)
            // Actually: framed_msg produces 32 + payload.len() bytes. When payload
            // is empty, total is 32, and echo checks `if len < 32 { return; }`.
            // len == 32 passes the check! So dest_ptr = ptr, payload_ptr = ptr+32,
            // payload_len = 0 → sends a 0-byte message.
            // Let's just check: if we got something, it matches.
            if let Some(received) = inbox.try_recv() {
                prop_assert_eq!(received.0, payload);
            }
        } else {
            let received = inbox.try_recv().expect("echo should return non-empty payload");
            prop_assert_eq!(received.0, payload);
        }
    }

    #[test]
    fn prop_double_always_sends_exactly_two_copies(payload in proptest::collection::vec(any::<u8>(), 1..500)) {
        let engine = SharedEngine::new().unwrap();
        let actor = WasmActorBuilder::new(engine, guest_wasm("double"))
            .build()
            .unwrap();

        let rt = Runtime::new(RuntimeConfig::default());
        let inbox = rt.new_inbox::<ByteMessage>().unwrap();
        let addr = rt.spawn(actor).unwrap();

        let msg = framed_msg(inbox.addr(), &payload);
        rt.send_to(addr, msg).unwrap();
        rt.tick();

        let first = inbox.try_recv().expect("double should send first copy");
        let second = inbox.try_recv().expect("double should send second copy");
        prop_assert_eq!(&first.0, &payload);
        prop_assert_eq!(&second.0, &payload);
        prop_assert!(inbox.try_recv().is_none(), "exactly two messages expected");
    }
}

// ── Alloc trap: unreachable in alloc, store must recover ─────────────────────

#[test]
fn alloc_traps_actor_survives_and_processes_next_message() {
    // Guest alloc traps on first call (counter=0), succeeds on subsequent calls.
    // The store must remain in a valid state after the alloc trap.
    let wat = r#"
        (module
            (import "swactor" "send" (func $send (param i32 i32 i32)))
            (memory (export "memory") 1)
            (global $counter (mut i32) (i32.const 0))

            (func (export "alloc") (param $size i32) (result i32)
                global.get $counter
                i32.const 1
                i32.add
                global.set $counter

                ;; First call: trap
                global.get $counter
                i32.const 1
                i32.eq
                if
                    unreachable
                end
                ;; Subsequent calls: return valid pointer
                i32.const 256
            )
            (func (export "handle") (param $ptr i32) (param $len i32)
                ;; Echo: send payload back to dest in first 32 bytes
                local.get $ptr
                local.get $ptr
                i32.const 32
                i32.add
                local.get $len
                i32.const 32
                i32.sub
                call $send
            )
        )
    "#;
    let wasm = wat::parse_str(wat).unwrap();
    let engine = SharedEngine::new().unwrap();
    let actor = WasmActorBuilder::new(engine, wasm).build().unwrap();

    let rt = Runtime::new(RuntimeConfig::default());
    let inbox = rt.new_inbox::<ByteMessage>().unwrap();
    let addr = rt.spawn(actor).unwrap();

    // First message: alloc traps → message dropped, no reply
    rt.send_to(addr, framed_msg(inbox.addr(), b"trap-in-alloc")).unwrap();
    rt.tick();
    assert!(inbox.try_recv().is_none(), "alloc trap should drop message");

    // Second message: alloc succeeds → echo should work
    let payload = b"after-alloc-trap";
    rt.send_to(addr, framed_msg(inbox.addr(), payload)).unwrap();
    rt.tick();
    let received = inbox.try_recv().expect("actor should recover after alloc trap");
    assert_eq!(received.0, payload);
}

// ── memory.grow during handle: guest expands memory, sends from new region ───

#[test]
fn memory_grow_during_handle_does_not_break_actor() {
    // Guest grows memory by 1 page during handle, then writes a value
    // into the new region and sends it. Verifies the host's Memory
    // handle tracks the new size.
    let wat = r#"
        (module
            (import "swactor" "send" (func $send (param i32 i32 i32)))
            (memory (export "memory") 1)

            (func (export "alloc") (param i32) (result i32)
                i32.const 256
            )
            (func (export "handle") (param $ptr i32) (param $len i32)
                ;; Grow memory by 1 page (64KiB → 128KiB)
                (drop (memory.grow (i32.const 1)))

                ;; Write marker byte into new page (offset 65536+100 = 65636)
                (i32.store8 (i32.const 65636) (i32.const 42))

                ;; Send: dest from first 32 bytes, payload from new region
                local.get $ptr    ;; dest_ptr
                i32.const 65636   ;; payload_ptr (in grown region)
                i32.const 1       ;; payload_len
                call $send
            )
        )
    "#;
    let wasm = wat::parse_str(wat).unwrap();
    let engine = SharedEngine::new().unwrap();
    let actor = WasmActorBuilder::new(engine, wasm).build().unwrap();

    let rt = Runtime::new(RuntimeConfig::default());
    let inbox = rt.new_inbox::<ByteMessage>().unwrap();
    let addr = rt.spawn(actor).unwrap();

    rt.send_to(addr, framed_msg(inbox.addr(), b"grow-test")).unwrap();
    rt.tick();

    let received = inbox.try_recv().expect("should receive from grown memory region");
    assert_eq!(received.0, vec![42], "payload should be the marker byte from new page");
}

// ── send overflow: dest_ptr near i32::MAX triggers checked_add overflow ──────

#[test]
fn send_with_dest_ptr_overflow_traps_actor_survives() {
    // Guest calls send with dest_ptr = i32::MAX (2147483647).
    // The host's checked_add(32) overflows → trap. Actor should survive.
    let wat = r#"
        (module
            (import "swactor" "send" (func $send (param i32 i32 i32)))
            (memory (export "memory") 1)

            (func (export "alloc") (param i32) (result i32)
                i32.const 256
            )
            (func (export "handle") (param i32 i32)
                i32.const 2147483647  ;; dest_ptr = i32::MAX
                i32.const 0           ;; payload_ptr
                i32.const 0           ;; payload_len
                call $send
            )
        )
    "#;
    let wasm = wat::parse_str(wat).unwrap();
    let engine = SharedEngine::new().unwrap();
    let actor = WasmActorBuilder::new(engine, wasm).build().unwrap();

    let rt = Runtime::new(RuntimeConfig::default());
    let addr = rt.spawn(actor).unwrap();

    rt.send_to(addr, ByteMessage(vec![1])).unwrap();
    rt.tick(); // send traps due to overflow — actor should survive

    // Verify actor is still alive
    rt.send_to(addr, ByteMessage(vec![2])).unwrap();
    rt.tick();
}

// ── Exact-fit allocation: ptr + len == memory size ───────────────────────────

#[test]
fn exact_fit_allocation_at_memory_boundary_succeeds() {
    // alloc returns 65536 - 10 = 65526. With a 10-byte message, the write
    // region is [65526..65536] — exactly fitting in 1 page. Should succeed.
    let wat = r#"
        (module
            (import "swactor" "send" (func $send (param i32 i32 i32)))
            (memory (export "memory") 1)

            (func (export "alloc") (param i32) (result i32)
                i32.const 65526  ;; 65536 - 10 = exact fit for 10-byte message
            )
            (func (export "handle") (param $ptr i32) (param $len i32)
                ;; Just echo: send everything back. But since alloc returns
                ;; 65526, the message was written to [65526..65536]. We need
                ;; to send from there. Use a fixed dest from offset 0 (zeroes).
                ;; Actually, the message was copied to ptr=65526 by the host.
                ;; We need the first 32 bytes as dest, but our message is only
                ;; 10 bytes. So handle gets (ptr=65526, len=10). With len < 32,
                ;; the echo guest would skip it. Let's just verify handle was
                ;; called by sending a known byte from offset 200.
                (i32.store8 (i32.const 200) (i32.const 99))
                ;; We can't easily echo from this offset, but we can verify
                ;; the handle was reached by using a global flag read in a
                ;; subsequent call.
            )
        )
    "#;
    let wasm = wat::parse_str(wat).unwrap();
    let engine = SharedEngine::new().unwrap();
    let actor = WasmActorBuilder::new(engine, wasm).build().unwrap();

    let rt = Runtime::new(RuntimeConfig::default());
    let addr = rt.spawn(actor).unwrap();

    // Send exactly 10 bytes — fits perfectly at ptr=65526
    rt.send_to(addr, ByteMessage(vec![0u8; 10])).unwrap();
    rt.tick(); // should NOT trigger OOB — exact fit

    // Actor survives — the bounds check passed
    rt.send_to(addr, ByteMessage(vec![1u8; 10])).unwrap();
    rt.tick();
}

// ── Off-by-one: alloc returns exactly memory size ────────────────────────────

#[test]
fn alloc_returns_exactly_memory_size_drops_message() {
    // alloc returns 65536 (exactly the size of 1-page memory).
    // Any non-zero length message means end > mem.len(), so it should be dropped.
    // For a zero-length message, ptr=65536, end=65536, which equals mem.len()
    // so end > mem.len() is false — that path technically works (no-op write).
    let wat = r#"
        (module
            (import "swactor" "send" (func $send (param i32 i32 i32)))
            (memory (export "memory") 1)
            (func (export "alloc") (param i32) (result i32)
                i32.const 65536  ;; exactly at memory boundary
            )
            (func (export "handle") (param i32 i32))
        )
    "#;
    let wasm = wat::parse_str(wat).unwrap();
    let engine = SharedEngine::new().unwrap();
    let actor = WasmActorBuilder::new(engine, wasm).build().unwrap();

    let rt = Runtime::new(RuntimeConfig::default());
    let addr = rt.spawn(actor).unwrap();

    // Non-zero message: end = 65536 + 5 = 65541 > 65536 → dropped
    rt.send_to(addr, ByteMessage(vec![0u8; 5])).unwrap();
    rt.tick();

    // Actor survives
    rt.send_to(addr, ByteMessage(vec![1u8; 5])).unwrap();
    rt.tick();
}

// ── Lifecycle: spawn and immediately stop without processing messages ────────

#[test]
fn spawn_and_stop_without_messages_is_clean() {
    // WASM actor spawned, immediately stopped, never processes a message.
    // The wasmtime Store should be dropped cleanly.
    let engine = SharedEngine::new().unwrap();
    let actor = WasmActorBuilder::new(engine, guest_wasm("echo"))
        .build()
        .unwrap();

    let rt = Runtime::new(RuntimeConfig::default());
    let addr = rt.spawn(actor).unwrap();

    // Stop immediately, no messages sent
    rt.stop_actor(addr).unwrap();
    rt.tick(); // process stop
    rt.tick(); // cleanup_dead

    // Actor is gone
    let result = rt.send_to(addr, ByteMessage(vec![1]));
    assert!(result.is_err(), "stopped actor should reject messages");
}

// ── 3-hop relay: WASM A → WASM B → WASM C → inbox ──────────────────────────

#[test]
fn three_hop_wasm_relay_delivers_final_payload() {
    // Three echo actors in sequence: A echoes to B, B echoes to C, C echoes to inbox.
    let engine = SharedEngine::new().unwrap();
    let wasm_bytes = guest_wasm("echo");

    let actor_a = WasmActorBuilder::new(engine.clone(), wasm_bytes.clone()).build().unwrap();
    let actor_b = WasmActorBuilder::new(engine.clone(), wasm_bytes.clone()).build().unwrap();
    let actor_c = WasmActorBuilder::new(engine, wasm_bytes).build().unwrap();

    let rt = Runtime::new(RuntimeConfig::default());
    let inbox = rt.new_inbox::<ByteMessage>().unwrap();

    let addr_a = rt.spawn(actor_a).unwrap();
    let addr_b = rt.spawn(actor_b).unwrap();
    let addr_c = rt.spawn(actor_c).unwrap();

    // Build nested framed message: A sends to B, B sends to C, C sends to inbox
    let final_payload = b"3-hops";
    let msg_for_c = framed_msg(inbox.addr(), final_payload);
    let msg_for_b = framed_msg(&addr_c, &msg_for_c.0);
    let msg_for_a = framed_msg(&addr_b, &msg_for_b.0);

    rt.send_to(addr_a, msg_for_a).unwrap();
    rt.tick(); // A → B
    rt.tick(); // B → C
    rt.tick(); // C → inbox

    let received = inbox.try_recv().expect("3-hop relay should deliver");
    assert_eq!(received.0, final_payload);
}

// ── memory.grow exhaustion: guest grows until failure ────────────────────────

#[test]
fn memory_grow_until_failure_actor_survives() {
    // Guest calls memory.grow repeatedly until it returns -1 (failure).
    // The actor should survive and the send should still work using
    // memory from before the failed grow.
    let wat = r#"
        (module
            (import "swactor" "send" (func $send (param i32 i32 i32)))
            (memory (export "memory") 1)

            (func (export "alloc") (param i32) (result i32)
                i32.const 256
            )
            (func (export "handle") (param $ptr i32) (param $len i32)
                (local $result i32)
                ;; Grow memory repeatedly until failure
                (block $done
                    (loop $grow
                        (local.set $result (memory.grow (i32.const 100)))
                        (br_if $done (i32.eq (local.get $result) (i32.const -1)))
                        (br $grow)
                    )
                )
                ;; After grow failure, write marker and send from original page
                (i32.store8 (i32.const 200) (i32.const 77))
                local.get $ptr    ;; dest_ptr
                i32.const 200     ;; payload_ptr
                i32.const 1       ;; payload_len
                call $send
            )
        )
    "#;
    let wasm = wat::parse_str(wat).unwrap();
    let engine = SharedEngine::new().unwrap();
    let actor = WasmActorBuilder::new(engine, wasm).build().unwrap();

    let rt = Runtime::new(RuntimeConfig::default());
    let inbox = rt.new_inbox::<ByteMessage>().unwrap();
    let addr = rt.spawn(actor).unwrap();

    rt.send_to(addr, framed_msg(inbox.addr(), b"grow-exhaust")).unwrap();
    rt.tick();

    let received = inbox.try_recv().expect("actor should work after grow exhaustion");
    assert_eq!(received.0, vec![77]);
}

// ── Stress: many WASM actors in a chain ──────────────────────────────────────

#[test]
fn ten_wasm_actors_chain_relay() {
    // 10 echo actors in a chain: actor[0]→actor[1]→...→actor[9]→inbox.
    // Tests that many WASM actors coexist and messages propagate through them.
    let engine = SharedEngine::new().unwrap();
    let wasm_bytes = guest_wasm("echo");

    let rt = Runtime::new(RuntimeConfig::default());
    let inbox = rt.new_inbox::<ByteMessage>().unwrap();

    let mut addrs = Vec::new();
    for _ in 0..10 {
        let actor = WasmActorBuilder::new(engine.clone(), wasm_bytes.clone())
            .build()
            .unwrap();
        addrs.push(rt.spawn(actor).unwrap());
    }

    // Build nested framed message from the inside out:
    // actor[9] receives [inbox_addr | final_payload] → echoes final_payload to inbox
    // actor[8] receives [addr[9] | msg_for_9] → echoes msg_for_9 to actor[9]
    // ...
    // actor[0] receives [addr[1] | msg_for_1] → echoes msg_for_1 to actor[1]
    let final_payload = b"chain-10";
    let mut msg = framed_msg(inbox.addr(), final_payload);
    for addr in addrs[1..].iter().rev() {
        msg = framed_msg(addr, &msg.0);
    }

    rt.send_to(addrs[0], msg).unwrap();
    for _ in 0..10 {
        rt.tick();
    }

    let received = inbox.try_recv().expect("10-actor chain should deliver");
    assert_eq!(received.0, final_payload);
}