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");
}

// ── 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");
    }
}