//! Single-worker, in-memory actor runtime. No durability or preemption guarantees.

use std::cell::RefCell;
use std::collections::{BTreeMap, VecDeque};
use std::num::NonZeroUsize;
use std::rc::Rc;
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};
use std::task::{Context, Wake, Waker};

use runa_actor::{
    ActorRef, ActorSnapshot, Envelope, Event, ExecutionSnapshot, MailboxId, Message, Status,
    SystemMessage,
};
use runa_code::CompiledArtifact;
use runa_core::{ImplementationId, LogEntry, LogLevel, Value};
use runa_rune::{
    ActorContext, ActorLogs, ExperimentResult, Invocation, Slice, value_from_rune, value_to_rune,
};

pub type RuntimeResult<T> = ExperimentResult<T>;

struct WakeFlag(AtomicBool);
impl Wake for WakeFlag {
    fn wake(self: Arc<Self>) {
        self.0.store(true, Ordering::SeqCst);
    }

    fn wake_by_ref(self: &Arc<Self>) {
        self.0.store(true, Ordering::SeqCst);
    }
}

struct Actor {
    snapshot: ActorSnapshot,
    mailbox: VecDeque<Envelope>,
    capacity: NonZeroUsize,
    active: Option<Invocation<rune::Value>>,
    context: ActorContext,
    reserved_sender: Option<ActorRef>,
    stopping: bool,
    shutdown: bool,
    pending_install: Option<ImplementationId>,
    wake: Arc<WakeFlag>,
    paused_from: Option<Status>,
}

#[derive(Default)]
pub struct Runtime {
    actors: BTreeMap<ActorRef, Actor>,
    implementations: BTreeMap<ImplementationId, Arc<CompiledArtifact>>,
    next_mailbox: u64,
    events: Vec<Event>,
    logs: Rc<RefCell<ActorLogs>>,
}

impl Runtime {
    pub fn publish(
        &mut self,
        artifact: Arc<CompiledArtifact>,
    ) -> ExperimentResult<ImplementationId> {
        let id = artifact.implementation_id;
        if let Some(existing) = self.implementations.get(&id) {
            if existing.source_id != artifact.source_id {
                return Err("implementation identity collision".into());
            }
            return Ok(id);
        }
        self.implementations.insert(id, artifact);
        Ok(id)
    }

    pub fn spawn(
        &mut self,
        implementation: ImplementationId,
        state: Value,
        capacity: NonZeroUsize,
    ) -> ExperimentResult<ActorRef> {
        if !self.implementations.contains_key(&implementation) {
            return Err("unknown implementation".into());
        }
        state.validate()?;
        self.next_mailbox = self
            .next_mailbox
            .checked_add(1)
            .ok_or("mailbox IDs exhausted")?;
        let mut actor_ref = ActorRef::new();
        while self.actors.contains_key(&actor_ref) {
            actor_ref = ActorRef::new();
        }
        self.actors.insert(
            actor_ref,
            Actor {
                snapshot: ActorSnapshot {
                    log_level: LogLevel::Info,
                    actor_ref,
                    mailbox: MailboxId(self.next_mailbox),
                    implementation,
                    incarnation: 1,
                    state,
                    status: Status::Idle,
                    execution: None,
                    queued: Vec::new(),
                    reserved: None,
                },
                mailbox: VecDeque::new(),
                capacity,
                active: None,
                context: ActorContext::with_logs(actor_ref, Rc::clone(&self.logs)),
                reserved_sender: None,
                stopping: false,
                shutdown: false,
                pending_install: None,
                wake: Arc::new(WakeFlag(AtomicBool::new(true))),
                paused_from: None,
            },
        );
        self.events.push(Event::Spawned(actor_ref));
        Ok(actor_ref)
    }

    pub fn send(&mut self, id: ActorRef, message: Message) -> ExperimentResult<()> {
        self.send_from(None, id, message)
    }

    pub fn send_from(
        &mut self,
        sender: Option<ActorRef>,
        id: ActorRef,
        message: Message,
    ) -> ExperimentResult<()> {
        message.validate()?;
        let actor = self.actors.get_mut(&id).ok_or("unknown actor")?;
        if actor.stopping || matches!(actor.snapshot.status, Status::Stopped | Status::Failed(_)) {
            return Err("actor is not accepting messages".into());
        }
        if actor.mailbox.len() + usize::from(actor.snapshot.reserved.is_some())
            >= actor.capacity.get()
        {
            return Err("mailbox full".into());
        }
        actor.mailbox.push_back(Envelope { sender, message });
        Ok(())
    }

    pub fn inspect(&self, id: ActorRef) -> ExperimentResult<ActorSnapshot> {
        let actor = self.actors.get(&id).ok_or("unknown actor")?;
        let mut snapshot = actor.snapshot.clone();
        snapshot.execution = actor.active.as_ref().map(|invocation| {
            let progress = invocation.progress();
            ExecutionSnapshot {
                vm_steps: progress.vm_steps,
                instruction_pointer: progress.instruction_pointer,
                call_depth: progress.call_depth,
            }
        });
        snapshot.queued = actor
            .mailbox
            .iter()
            .map(|envelope| envelope.message.clone())
            .collect();
        Ok(snapshot)
    }

    pub fn edit_actor(
        &mut self,
        id: ActorRef,
        state: Option<Value>,
        messages: Option<(Option<Message>, Vec<Message>)>,
    ) -> ExperimentResult<()> {
        let actor = self.actors.get_mut(&id).ok_or("unknown actor")?;
        if actor.active.is_some()
            || !matches!(
                actor.snapshot.status,
                Status::Paused | Status::Stopped | Status::Failed(_)
            )
        {
            return Err(
                "editing requires a paused, stopped, or failed actor with no active invocation"
                    .into(),
            );
        }
        if let Some(state) = &state {
            state.validate()?;
        }
        if let Some((reserved, queued)) = &messages {
            for message in reserved.iter().chain(queued) {
                message.validate()?;
            }
            if queued.len() + usize::from(reserved.is_some()) > actor.capacity.get() {
                return Err("mailbox full".into());
            }
        }
        if let Some(state) = state {
            actor.snapshot.state = state;
        }
        let Some((reserved, queued)) = messages else {
            return Ok(());
        };
        // Preserve sender metadata for unchanged messages, including the failed reservation.
        if actor.snapshot.reserved != reserved {
            actor.reserved_sender = None;
        }
        actor.snapshot.reserved = reserved;
        let previous = std::mem::take(&mut actor.mailbox);
        actor.mailbox = queued
            .into_iter()
            .enumerate()
            .map(|(index, message)| {
                let sender = previous
                    .get(index)
                    .filter(|old| old.message == message)
                    .and_then(|old| old.sender);
                Envelope { sender, message }
            })
            .collect();
        Ok(())
    }

    pub fn set_log_level(&mut self, id: ActorRef, level: LogLevel) -> ExperimentResult<()> {
        self.actors
            .get_mut(&id)
            .ok_or("unknown actor")?
            .snapshot
            .log_level = level;
        self.logs.borrow_mut().set_level(id, level);
        Ok(())
    }

    pub fn log_level(&self, id: ActorRef) -> ExperimentResult<LogLevel> {
        self.actors.get(&id).ok_or("unknown actor")?;
        Ok(self.logs.borrow().level(id))
    }

    pub fn logs(&self, actors: &[ActorRef], after: u64) -> ExperimentResult<Vec<LogEntry>> {
        for actor in actors {
            self.actors.get(actor).ok_or("unknown actor")?;
        }
        Ok(self.logs.borrow().entries(actors, after))
    }

    pub fn set_log_sink(&mut self, sink: impl Fn(LogEntry) + 'static) {
        self.logs.borrow_mut().set_sink(sink);
    }

    pub fn actors(&self) -> Vec<ActorSnapshot> {
        self.actors
            .values()
            .map(|actor| {
                let mut snapshot = actor.snapshot.clone();
                snapshot.execution = actor.active.as_ref().map(|invocation| {
                    let progress = invocation.progress();
                    ExecutionSnapshot {
                        vm_steps: progress.vm_steps,
                        instruction_pointer: progress.instruction_pointer,
                        call_depth: progress.call_depth,
                    }
                });
                snapshot.queued = actor
                    .mailbox
                    .iter()
                    .map(|envelope| envelope.message.clone())
                    .collect();
                snapshot
            })
            .collect()
    }

    pub fn stop(&mut self, id: ActorRef) -> ExperimentResult<()> {
        let actor = self.actors.get_mut(&id).ok_or("unknown actor")?;
        if matches!(actor.snapshot.status, Status::Stopped | Status::Failed(_)) {
            return Err("actor is not running".into());
        }
        actor.stopping = true;
        if let Some(status) = actor.paused_from.take() {
            actor.snapshot.status = status;
        }
        Ok(())
    }

    pub fn kill(&mut self, id: ActorRef) -> ExperimentResult<()> {
        let actor = self.actors.get_mut(&id).ok_or("unknown actor")?;
        actor.active = None;
        actor.context.take_effects();
        actor.stopping = false;
        actor.shutdown = false;
        actor.pending_install = None;
        actor.paused_from = None;
        actor.snapshot.status = Status::Stopped;
        self.events.push(Event::Stopped(id));
        Ok(())
    }

    pub fn system(&mut self, id: ActorRef, message: SystemMessage) -> ExperimentResult<()> {
        match message {
            SystemMessage::Stop => self.stop(id),
            SystemMessage::Kill => self.kill(id),
            SystemMessage::Restart => {
                self.kill(id)?;
                self.restart(id)
            }
            SystemMessage::Pause => self.pause(id),
            SystemMessage::Resume => self.resume(id),
            SystemMessage::Reload(implementation) | SystemMessage::Become(implementation) => {
                self.schedule_install(id, implementation)
            }
        }
    }

    pub fn schedule_install(
        &mut self,
        id: ActorRef,
        implementation: ImplementationId,
    ) -> ExperimentResult<()> {
        if !self.implementations.contains_key(&implementation) {
            return Err("unknown implementation".into());
        }
        let actor = self.actors.get_mut(&id).ok_or("unknown actor")?;
        if actor.stopping || matches!(actor.snapshot.status, Status::Stopped | Status::Failed(_)) {
            return Err("actor is not running".into());
        }
        actor.pending_install = Some(implementation);
        Ok(())
    }

    /// Pause scheduling between worker ticks without discarding an invocation.
    pub fn pause(&mut self, id: ActorRef) -> ExperimentResult<()> {
        let actor = self.actors.get_mut(&id).ok_or("unknown actor")?;
        if matches!(
            actor.snapshot.status,
            Status::Paused | Status::Stopped | Status::Failed(_)
        ) {
            return Err("actor cannot be paused in its current status".into());
        }
        actor.paused_from = Some(actor.snapshot.status.clone());
        actor.snapshot.status = Status::Paused;
        self.events.push(Event::Paused(id));
        Ok(())
    }

    /// Resume from the exact runnable or awaiting state saved by `pause`.
    pub fn resume(&mut self, id: ActorRef) -> ExperimentResult<()> {
        let actor = self.actors.get_mut(&id).ok_or("unknown actor")?;
        let status = actor.paused_from.take().ok_or("actor is not paused")?;
        actor.snapshot.status = status;
        self.events.push(Event::Resumed(id));
        Ok(())
    }

    /// Cancel an active handler and requeue its reserved message for retry.
    ///
    /// This does not roll back side effects already performed by native code.
    pub fn cancel(&mut self, id: ActorRef) -> ExperimentResult<()> {
        let actor = self.actors.get_mut(&id).ok_or("unknown actor")?;
        if actor.active.is_none() {
            return Err("actor has no active invocation to cancel".into());
        }
        let message = actor
            .snapshot
            .reserved
            .take()
            .ok_or("active invocation has no reserved message")?;
        actor.active = None;
        actor.paused_from = None;
        actor.context.take_effects();
        actor.mailbox.push_front(Envelope {
            sender: actor.reserved_sender.take(),
            message,
        });
        actor.snapshot.status = Status::Idle;
        self.events.push(Event::InvocationCancelled(id));
        Ok(())
    }

    pub fn restart(&mut self, id: ActorRef) -> ExperimentResult<()> {
        let actor = self.actors.get_mut(&id).ok_or("unknown actor")?;
        if !matches!(actor.snapshot.status, Status::Stopped | Status::Failed(_)) {
            return Err("stop the actor before restarting".into());
        }
        actor.snapshot.incarnation = actor
            .snapshot
            .incarnation
            .checked_add(1)
            .ok_or("incarnation exhausted")?;
        if let Some(message) = actor.snapshot.reserved.take() {
            actor.mailbox.push_front(Envelope {
                sender: actor.reserved_sender.take(),
                message,
            });
        }
        actor.snapshot.status = Status::Idle;
        actor.paused_from = None;
        self.events.push(Event::Restarted(id));
        Ok(())
    }

    /// Install an immutable implementation between handler invocations.
    pub fn install(
        &mut self,
        id: ActorRef,
        implementation: ImplementationId,
    ) -> ExperimentResult<()> {
        self.install_with_migration(id, implementation, Ok)
    }

    /// Migrate committed state and install an implementation as one safe-point
    /// operation. A migration error leaves both unchanged.
    pub fn install_with_migration(
        &mut self,
        id: ActorRef,
        implementation: ImplementationId,
        migrate: impl FnOnce(Value) -> ExperimentResult<Value>,
    ) -> ExperimentResult<()> {
        if !self.implementations.contains_key(&implementation) {
            return Err("unknown implementation".into());
        }
        let current = self.actors.get(&id).ok_or("unknown actor")?;
        if current.active.is_some()
            || current.snapshot.reserved.is_some()
            || current.snapshot.status != Status::Idle
        {
            return Err("behavior can only be installed at an idle handler boundary".into());
        }
        let previous_state = current.snapshot.state.clone();
        let migrated_state = migrate(previous_state.clone())
            .map_err(|error| format!("state migration failed: {error}"))?;
        migrated_state.validate()?;

        let actor = self.actors.get_mut(&id).expect("actor checked above");
        actor.snapshot.state = migrated_state.clone();
        actor.snapshot.implementation = implementation;
        self.events
            .push(Event::BehaviorInstalled(id, implementation));
        if migrated_state != previous_state {
            self.events
                .push(Event::StateMigrated(id, previous_state, migrated_state));
        }
        Ok(())
    }

    pub fn drain_events(&mut self) -> Vec<Event> {
        std::mem::take(&mut self.events)
    }

    fn fail(&mut self, id: &ActorRef, error: String) {
        let actor = self.actors.get_mut(id).expect("scheduled actor exists");
        actor.active = None;
        actor.context.take_effects();
        actor.snapshot.status = Status::Failed(error.clone());
        self.events.push(Event::Failed(*id, error));
    }

    fn commit(&mut self, id: &ActorRef, state: rune::Value) -> ExperimentResult<()> {
        let state = value_from_rune(&state)?;
        let effects = self.actors[id].context.take_effects();
        let mut counts = BTreeMap::<ActorRef, usize>::new();
        for effect in &effects {
            let target = &effect.recipient;
            let actor = self
                .actors
                .get(target)
                .ok_or("effect targets unknown actor")?;
            if (actor.stopping && target != id)
                || matches!(actor.snapshot.status, Status::Stopped | Status::Failed(_))
            {
                return Err("effect targets unavailable actor".into());
            }
            let count = counts.entry(*target).or_default();
            *count += 1;
            let retained = usize::from(target != id && actor.snapshot.reserved.is_some());
            if actor.mailbox.len() + retained + *count > actor.capacity.get() {
                return Err("effect exceeds mailbox capacity".into());
            }
        }
        let actor = self.actors.get_mut(id).expect("scheduled actor exists");
        actor.snapshot.state = state;
        actor.snapshot.reserved = None;
        actor.snapshot.status = Status::Idle;
        actor.active = None;
        actor.reserved_sender = None;
        if actor.shutdown {
            actor.shutdown = false;
            actor.stopping = false;
            actor.snapshot.status = Status::Stopped;
            self.events.push(Event::Stopped(*id));
        }
        for effect in effects {
            self.actors
                .get_mut(&effect.recipient)
                .expect("validated target")
                .mailbox
                .push_back(Envelope {
                    sender: effect.sender,
                    message: effect.message,
                });
        }
        self.events.push(Event::Committed(*id));
        Ok(())
    }

    /// Give each runnable actor one bounded slice. Awaiting actors require a wake.
    pub fn tick(&mut self, budget: NonZeroUsize) {
        let ids: Vec<_> = self.actors.keys().copied().collect();
        for id in ids {
            let actor = self.actors.get_mut(&id).expect("scheduled actor exists");
            if matches!(
                actor.snapshot.status,
                Status::Paused | Status::Stopped | Status::Failed(_)
            ) {
                continue;
            }
            if actor.active.is_none() {
                if actor.stopping {
                    let program = self.implementations[&actor.snapshot.implementation].program();
                    if !program.has_function("on_shutdown") {
                        actor.snapshot.status = Status::Stopped;
                        actor.stopping = false;
                        self.events.push(Event::Stopped(id));
                        continue;
                    }
                    let invocation = (|| -> ExperimentResult<_> {
                        Ok(Invocation::start(
                            program.vm(),
                            "on_shutdown",
                            (value_to_rune(&actor.snapshot.state)?,),
                        )?)
                    })();
                    match invocation {
                        Ok(invocation) => {
                            actor.active = Some(invocation);
                            actor.shutdown = true;
                            actor.context = ActorContext::with_logs(id, Rc::clone(&self.logs));
                        }
                        Err(error) => {
                            self.fail(&id, error.to_string());
                            continue;
                        }
                    }
                } else {
                    if let Some(implementation) = actor.pending_install.take() {
                        actor.snapshot.implementation = implementation;
                        self.events
                            .push(Event::BehaviorInstalled(id, implementation));
                    }
                    let Some(message) = actor.mailbox.pop_front() else {
                        continue;
                    };
                    actor.snapshot.reserved = Some(message.message.clone());
                    actor.reserved_sender = message.sender;
                    actor.context = ActorContext::with_logs(id, Rc::clone(&self.logs));
                    let program = self.implementations[&actor.snapshot.implementation].program();
                    let invocation = (|| -> ExperimentResult<_> {
                        let state = value_to_rune(&actor.snapshot.state)?;
                        let sender =
                            value_to_rune(&message.sender.map_or(Value::Null, Value::ActorRef))?;
                        let message = value_to_rune(&message.message)?;
                        Ok(Invocation::start(
                            program.vm(),
                            "handle",
                            (actor.context.clone(), state, sender, message),
                        )?)
                    })();
                    match invocation {
                        Ok(invocation) => actor.active = Some(invocation),
                        Err(error) => {
                            self.fail(&id, error.to_string());
                            continue;
                        }
                    }
                }
                actor.snapshot.status = Status::Running;
            }
            if actor.snapshot.status == Status::Awaiting && !actor.wake.0.load(Ordering::SeqCst) {
                continue;
            }
            actor.wake.0.store(false, Ordering::SeqCst);
            let waker = Waker::from(Arc::clone(&actor.wake));
            let mut cx = Context::from_waker(&waker);
            let outcome = actor
                .active
                .as_mut()
                .expect("active invocation")
                .poll_slice(&mut cx, budget);
            match outcome {
                Slice::Completed(transition) => {
                    if let Err(error) = self.commit(&id, transition) {
                        self.fail(&id, error.to_string());
                    }
                }
                Slice::Awaiting => actor.snapshot.status = Status::Awaiting,
                Slice::BudgetExhausted => actor.snapshot.status = Status::Running,
                Slice::Failed(error) => self.fail(&id, error.to_string()),
            }
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use runa_core::MAX_VALUE_NODES;

    #[test]
    fn editing_is_guarded_atomic_and_can_repair_a_failed_reservation() -> ExperimentResult<()> {
        let mut runtime = Runtime::default();
        let implementation = publish(
            &mut runtime,
            "pub fn handle(ctx, state, sender, message) { state + message.value }",
        )?;
        let actor = runtime.spawn(implementation, Value::Integer(1), capacity(2))?;
        assert!(
            runtime
                .edit_actor(actor, Some(Value::Integer(9)), None)
                .is_err()
        );
        runtime.send(actor, Value::Null)?;
        runtime.send(actor, Value::Integer(4))?;
        for _ in 0..10 {
            runtime.tick(capacity(32));
        }
        assert!(matches!(runtime.inspect(actor)?.status, Status::Failed(_)));
        assert!(
            runtime
                .edit_actor(
                    actor,
                    Some(Value::Integer(9)),
                    Some((Some(Value::Null), vec![Value::Null, Value::Null]))
                )
                .is_err()
        );
        assert_eq!(runtime.inspect(actor)?.state, Value::Integer(1));
        runtime.edit_actor(actor, Some(Value::Integer(10)), None)?;
        assert_eq!(runtime.inspect(actor)?.queued, vec![Value::Integer(4)]);
        let message = Value::Record(BTreeMap::from([("value".to_owned(), Value::Integer(2))]));
        runtime.edit_actor(actor, None, Some((Some(message), vec![])))?;
        runtime.restart(actor)?;
        for _ in 0..10 {
            runtime.tick(capacity(32));
        }
        assert_eq!(runtime.inspect(actor)?.state, Value::Integer(12));
        runtime.pause(actor)?;
        runtime.edit_actor(
            actor,
            Some(Value::Integer(20)),
            Some((None, vec![Value::Null])),
        )?;
        assert_eq!(runtime.inspect(actor)?.status, Status::Paused);
        runtime.resume(actor)?;
        let runaway = publish(
            &mut runtime,
            "pub fn handle(ctx, state, sender, message) { loop {} }",
        )?;
        let active = runtime.spawn(runaway, Value::Null, capacity(2))?;
        runtime.send(active, Value::Null)?;
        runtime.tick(capacity(1));
        runtime.pause(active)?;
        assert!(
            runtime
                .edit_actor(active, Some(Value::Integer(9)), Some((None, vec![])))
                .is_err()
        );
        runtime.kill(active)?;
        runtime.edit_actor(active, Some(Value::Integer(9)), Some((None, vec![])))?;
        Ok(())
    }

    #[test]
    fn logs_are_immediate_filtered_ordered_bounded_and_survive_failure() -> ExperimentResult<()> {
        let mut runtime = Runtime::default();
        let emitted = Rc::new(RefCell::new(Vec::new()));
        let sink = emitted.clone();
        runtime.set_log_sink(move |entry| sink.borrow_mut().push(entry));
        let implementation = publish(
            &mut runtime,
            r#"pub fn handle(ctx, state, sender, message) {
                ctx.trace("trace"); ctx.debug("debug"); ctx.info("info");
                ctx.warn("warn"); ctx.error("error"); ctx.fatal("fatal"); ctx.log("info", "generic");
                panic("failed after logging")
            }"#,
        )?;
        let actor = runtime.spawn(implementation, Value::Integer(1), capacity(2))?;
        assert_eq!(runtime.log_level(actor)?, LogLevel::Info);
        runtime.send(actor, Value::Null)?;
        for _ in 0..10 {
            runtime.tick(capacity(32));
        }
        let entries = runtime.logs(&[actor], 0)?;
        assert_eq!(
            entries
                .iter()
                .map(|entry| entry.message.as_str())
                .collect::<Vec<_>>(),
            ["info", "warn", "error", "fatal", "generic"]
        );
        assert!(matches!(runtime.inspect(actor)?.status, Status::Failed(_)));
        runtime.set_log_level(actor, LogLevel::Fatal)?;
        runtime.restart(actor)?;
        for _ in 0..10 {
            runtime.tick(capacity(32));
        }
        assert_eq!(
            runtime
                .logs(&[actor], entries.last().unwrap().sequence)?
                .len(),
            1
        );
        let busy = publish(
            &mut runtime,
            r#"pub fn handle(ctx, state, sender, message) {
            let n = 0; while n < 1005 { ctx.info("entry"); n += 1; } loop {}
        }"#,
        )?;
        let second = runtime.spawn(busy, Value::Null, capacity(2))?;
        runtime.send(second, Value::Null)?;
        for _ in 0..100 {
            runtime.tick(capacity(1000));
        }
        let retained = runtime.logs(&[second], 0)?;
        assert_eq!(retained.len(), 1000);
        assert!(
            retained
                .windows(2)
                .all(|pair| pair[0].sequence < pair[1].sequence)
        );
        runtime.kill(second)?;
        assert_eq!(runtime.logs(&[second], 0)?.len(), 1000);
        assert_eq!(runtime.logs(&[], 0)?.len(), 1006);
        assert_eq!(emitted.borrow().len(), 1011);
        assert_eq!(emitted.borrow().last(), retained.last());
        Ok(())
    }

    #[test]
    fn context_sends_copy_at_call_and_are_discarded_on_kill() -> ExperimentResult<()> {
        let mut runtime = Runtime::default();
        let receiver_code = publish(
            &mut runtime,
            "pub fn handle(ctx, state, sender, message) { message }",
        )?;
        let receiver = runtime.spawn(receiver_code, Value::Null, capacity(4))?;
        let code = publish(
            &mut runtime,
            "pub fn handle(ctx, state, sender, message) { let data = #{items: []}; ctx.send(state, data); data.items.push(7); data }",
        )?;
        let actor = runtime.spawn(code, Value::ActorRef(receiver), capacity(2))?;
        runtime.send(actor, Value::Null)?;
        for _ in 0..100 {
            runtime.tick(capacity(32));
        }
        assert_eq!(
            runtime.inspect(receiver)?.state,
            Value::Record(BTreeMap::from([("items".to_owned(), Value::List(vec![])),]))
        );
        let runaway = publish(
            &mut runtime,
            "pub fn handle(ctx, state, sender, message) { ctx.send(state, 999); loop {} }",
        )?;
        let killed = runtime.spawn(runaway, Value::ActorRef(receiver), capacity(2))?;
        runtime.send(killed, Value::Null)?;
        for _ in 0..100 {
            runtime.tick(capacity(32));
        }
        runtime.system(killed, SystemMessage::Kill)?;
        for _ in 0..100 {
            runtime.tick(capacity(32));
        }
        assert_eq!(
            runtime.inspect(receiver)?.state,
            Value::Record(BTreeMap::from([("items".to_owned(), Value::List(vec![])),]))
        );
        Ok(())
    }

    #[test]
    fn sender_envelopes_support_replies_and_forwarding_without_message_metadata()
    -> ExperimentResult<()> {
        let mut runtime = Runtime::default();
        let sink = publish(
            &mut runtime,
            "pub fn handle(ctx, state, sender, message) { #{sender: sender, data: message} }",
        )?;
        let forward = publish(
            &mut runtime,
            "pub fn handle(ctx, state, sender, message) { ctx.send_from(sender, state, message); state }",
        )?;
        let origin = publish(
            &mut runtime,
            "pub fn handle(ctx, state, sender, message) { ctx.send(state, message); state }",
        )?;
        let sink_ref = runtime.spawn(sink, Value::Null, capacity(4))?;
        let forward_ref = runtime.spawn(forward, Value::ActorRef(sink_ref), capacity(4))?;
        let origin_ref = runtime.spawn(origin, Value::ActorRef(forward_ref), capacity(4))?;
        runtime.send(origin_ref, Value::String("stop".to_owned()))?;
        for _ in 0..100 {
            runtime.tick(capacity(32));
        }
        assert_eq!(
            runtime.inspect(sink_ref)?.state,
            Value::Record(BTreeMap::from([
                ("sender".to_owned(), Value::ActorRef(origin_ref)),
                ("data".to_owned(), Value::String("stop".to_owned())),
            ]))
        );
        assert_eq!(runtime.inspect(sink_ref)?.status, Status::Idle);
        let reply = publish(
            &mut runtime,
            "pub fn handle(ctx, state, sender, message) { ctx.send(sender, message); state }",
        )?;
        let reply_ref = runtime.spawn(reply, Value::Null, capacity(4))?;
        runtime.send_from(Some(sink_ref), reply_ref, Value::Integer(42))?;
        for _ in 0..100 {
            runtime.tick(capacity(32));
        }
        assert_eq!(
            runtime.inspect(sink_ref)?.state,
            Value::Record(BTreeMap::from([
                ("sender".to_owned(), Value::ActorRef(reply_ref)),
                ("data".to_owned(), Value::Integer(42)),
            ]))
        );
        Ok(())
    }

    #[test]
    fn graceful_stop_finishes_handler_then_runs_shutdown_without_draining_mailbox()
    -> ExperimentResult<()> {
        let mut runtime = Runtime::default();
        let implementation = publish(
            &mut runtime,
            "pub fn handle(ctx, state, sender, message) { let n = 0; while n < 20 { n += 1; } state + message } pub fn on_shutdown(state) { state + 100 }",
        )?;
        let actor = runtime.spawn(implementation, Value::Integer(0), capacity(3))?;
        runtime.send(actor, Value::Integer(1))?;
        runtime.send(actor, Value::Integer(2))?;
        runtime.tick(capacity(1));
        runtime.system(actor, SystemMessage::Stop)?;
        assert!(runtime.send(actor, Value::Null).is_err());
        for _ in 0..200 {
            runtime.tick(capacity(32));
        }
        let snapshot = runtime.inspect(actor)?;
        assert_eq!(snapshot.state, Value::Integer(101), "{snapshot:?}");
        assert_eq!(snapshot.status, Status::Stopped);
        assert_eq!(snapshot.queued, vec![Value::Integer(2)]);
        assert_eq!(snapshot.reserved, None);
        runtime.system(actor, SystemMessage::Restart)?;
        for _ in 0..200 {
            runtime.tick(capacity(32));
        }
        assert_eq!(runtime.inspect(actor)?.state, Value::Integer(103));
        let self_send = publish(
            &mut runtime,
            "pub fn handle(ctx, state, sender, message) { ctx.send(message, 42); state + 1 } pub fn on_shutdown(state) { state + 100 }",
        )?;
        let actor = runtime.spawn(self_send, Value::Integer(0), capacity(2))?;
        runtime.send(actor, Value::ActorRef(actor))?;
        runtime.tick(capacity(1));
        runtime.system(actor, SystemMessage::Stop)?;
        for _ in 0..100 {
            runtime.tick(capacity(32));
        }
        let snapshot = runtime.inspect(actor)?;
        assert_eq!(snapshot.state, Value::Integer(101));
        assert_eq!(snapshot.status, Status::Stopped);
        assert_eq!(snapshot.queued, vec![Value::Integer(42)]);
        Ok(())
    }

    #[test]
    fn kill_skips_shutdown_and_reload_waits_for_active_handler() -> ExperimentResult<()> {
        let mut runtime = Runtime::default();
        let first = publish(
            &mut runtime,
            "pub fn handle(ctx, state, sender, message) { state + message } pub fn on_shutdown(state) { panic(\"must not run\") }",
        )?;
        let second = publish(
            &mut runtime,
            "pub fn handle(ctx, state, sender, message) { state + message * 10 }",
        )?;
        let actor = runtime.spawn(first, Value::Integer(0), capacity(3))?;
        runtime.send(actor, Value::Integer(1))?;
        runtime.send(actor, Value::Integer(2))?;
        runtime.tick(capacity(1));
        runtime.system(actor, SystemMessage::Reload(second))?;
        for _ in 0..100 {
            runtime.tick(capacity(32));
        }
        assert_eq!(runtime.inspect(actor)?.state, Value::Integer(21));
        runtime.system(actor, SystemMessage::Kill)?;
        assert_eq!(runtime.inspect(actor)?.status, Status::Stopped);
        assert_eq!(runtime.inspect(actor)?.state, Value::Integer(21));
        runtime.system(actor, SystemMessage::Restart)?;
        assert_eq!(runtime.inspect(actor)?.implementation, second);
        Ok(())
    }

    #[test]
    fn absent_and_failing_shutdown_hooks_are_explicit() -> ExperimentResult<()> {
        let mut runtime = Runtime::default();
        for (hook, failed) in [
            ("", false),
            (
                "pub fn on_shutdown(state) { panic(\"shutdown failed\") }",
                true,
            ),
        ] {
            let implementation = publish(
                &mut runtime,
                &format!("pub fn handle(ctx, state, sender, message) {{ state }} {hook}"),
            )?;
            let actor = runtime.spawn(implementation, Value::Integer(7), capacity(1))?;
            runtime.system(actor, SystemMessage::Stop)?;
            for _ in 0..100 {
                runtime.tick(capacity(32));
            }
            let snapshot = runtime.inspect(actor)?;
            assert_eq!(snapshot.state, Value::Integer(7));
            if failed {
                assert!(matches!(snapshot.status, Status::Failed(_)));
            } else {
                assert_eq!(snapshot.status, Status::Stopped);
            }
        }
        Ok(())
    }

    fn test_message(kind: i64, value: i64, target: Option<runa_core::ActorRef>) -> Value {
        Value::Record(std::collections::BTreeMap::from([
            ("kind".to_owned(), Value::Integer(kind)),
            ("value".to_owned(), Value::Integer(value)),
            (
                "target".to_owned(),
                target.map_or(Value::Null, Value::ActorRef),
            ),
        ]))
    }

    use runa_code::CodeStore;

    #[test]
    fn nested_state_and_messages_are_private_and_sends_copy_owned_values() -> ExperimentResult<()> {
        let mut runtime = Runtime::default();
        let receiver_code = publish(
            &mut runtime,
            "pub fn handle(ctx, state, sender, message) { message.items.push(9); message }",
        )?;
        let sender_code = publish(
            &mut runtime,
            "pub fn handle(ctx, state, sender, message) { state.items.push(message.text); ctx.send(message.target, state); state }",
        )?;
        let initial = Value::Record(BTreeMap::from([(
            "items".to_owned(),
            Value::List(vec![Value::Bool(true), Value::Null]),
        )]));
        let receiver = runtime.spawn(receiver_code, Value::Null, capacity(2))?;
        let sender = runtime.spawn(sender_code, initial.clone(), capacity(2))?;
        let message = Value::Record(BTreeMap::from([
            ("target".to_owned(), Value::ActorRef(receiver)),
            ("text".to_owned(), Value::String("hello world".to_owned())),
        ]));
        runtime.send(sender, message.clone())?;
        runtime.tick(capacity(1));
        assert_eq!(runtime.inspect(sender)?.state, initial);
        assert_eq!(runtime.inspect(sender)?.reserved, Some(message));
        for _ in 0..100 {
            runtime.tick(capacity(32));
        }
        let sender_state = Value::Record(BTreeMap::from([(
            "items".to_owned(),
            Value::List(vec![
                Value::Bool(true),
                Value::Null,
                Value::String("hello world".to_owned()),
            ]),
        )]));
        let receiver_state = Value::Record(BTreeMap::from([(
            "items".to_owned(),
            Value::List(vec![
                Value::Bool(true),
                Value::Null,
                Value::String("hello world".to_owned()),
                Value::Integer(9),
            ]),
        )]));
        assert_eq!(runtime.inspect(sender)?.state, sender_state);
        assert_eq!(runtime.inspect(receiver)?.state, receiver_state);
        Ok(())
    }

    #[test]
    fn mutated_drafts_and_invalid_structured_effects_never_partially_commit() -> ExperimentResult<()>
    {
        for source in [
            "pub fn handle(ctx, state, sender, message) { state.items.push(2); panic(\"failed\"); state }",
            "pub fn handle(ctx, state, sender, message) { state.items.push(2); ctx.send(message, state); ctx.send(\"not a reference\", state); state }",
            "pub fn handle(ctx, state, sender, message) { state.items.push(2); ctx.send(message, state); ctx.send(message, 1.5); state }",
            "pub fn handle(ctx, state, sender, message) { state.items.push(2); ctx.send(message, state); Some(state) }",
            "pub fn handle(ctx, state, sender, message) { state.items.push(state); state }",
        ] {
            let mut runtime = Runtime::default();
            let code = publish(&mut runtime, source)?;
            let initial = Value::Record(BTreeMap::from([(
                "items".to_owned(),
                Value::List(vec![Value::Integer(1)]),
            )]));
            let actor = runtime.spawn(code, initial.clone(), capacity(4))?;
            runtime.send(actor, Value::ActorRef(actor))?;
            for _ in 0..100 {
                runtime.tick(capacity(32));
            }
            let snapshot = runtime.inspect(actor)?;
            assert_eq!(snapshot.state, initial, "{source}");
            assert_eq!(snapshot.reserved, Some(Value::ActorRef(actor)), "{source}");
            assert!(snapshot.queued.is_empty(), "{source}");
            assert!(matches!(snapshot.status, Status::Failed(_)), "{source}");
        }
        Ok(())
    }

    #[test]
    fn invalid_input_and_migration_values_leave_the_image_unchanged() -> ExperimentResult<()> {
        let mut runtime = Runtime::default();
        let code = publish(
            &mut runtime,
            "pub fn handle(ctx, state, sender, message) { message }",
        )?;
        let invalid = Value::List(vec![Value::Null; MAX_VALUE_NODES]);
        assert!(runtime.spawn(code, invalid.clone(), capacity(2)).is_err());
        assert!(runtime.actors().is_empty());
        let actor = runtime.spawn(code, Value::Bool(true), capacity(2))?;
        assert!(runtime.send(actor, invalid.clone()).is_err());
        assert!(
            runtime
                .install_with_migration(actor, code, |_| Ok(invalid))
                .is_err()
        );
        let snapshot = runtime.inspect(actor)?;
        assert_eq!(snapshot.state, Value::Bool(true));
        assert!(snapshot.queued.is_empty());
        Ok(())
    }

    fn capacity(n: usize) -> NonZeroUsize {
        NonZeroUsize::new(n).unwrap()
    }

    fn publish(runtime: &mut Runtime, source: &str) -> ExperimentResult<ImplementationId> {
        runtime.publish(CodeStore::default().compile(source)?)
    }

    #[test]
    fn two_actors_isolation_and_effects() -> ExperimentResult<()> {
        let mut runtime = Runtime::default();
        let code = publish(
            &mut runtime,
            "pub fn handle(ctx, state, sender, message) { if message.kind == 1 { ctx.send(message.target, #{kind: 0, value: 5, target: ()}); state + 1 } else { state + message.value } }",
        )?;
        let first = runtime.spawn(code, Value::Integer(0), capacity(4))?;
        let second = runtime.spawn(code, Value::Integer(10), capacity(4))?;
        runtime.send(first, test_message(1, 0, Some(second)))?;
        for _ in 0..100 {
            runtime.tick(capacity(10));
        }
        assert_eq!(runtime.inspect(first)?.state, Value::Integer(1));
        assert_eq!(runtime.inspect(second)?.state, Value::Integer(15));
        assert!(runtime.inspect(first)?.reserved.is_none());
        assert!(
            runtime
                .drain_events()
                .iter()
                .any(|e| matches!(e, Event::Committed(id) if *id == second))
        );
        Ok(())
    }

    #[test]
    fn runaway_does_not_starve_healthy_actor() -> ExperimentResult<()> {
        let mut runtime = Runtime::default();
        let runaway = publish(
            &mut runtime,
            "pub fn handle(ctx, state, sender, message) { loop {} }",
        )?;
        let healthy = publish(
            &mut runtime,
            "pub fn handle(ctx, state, sender, message) { state + message.value }",
        )?;
        let first = runtime.spawn(runaway, Value::Integer(0), capacity(1))?;
        let second = runtime.spawn(healthy, Value::Integer(0), capacity(2))?;
        runtime.send(first, test_message(0, 1, None))?;
        runtime.send(second, test_message(0, 42, None))?;
        for _ in 0..100 {
            runtime.tick(capacity(5));
        }
        assert_eq!(runtime.inspect(first)?.status, Status::Running);
        assert_eq!(runtime.inspect(first)?.state, Value::Integer(0));
        assert_eq!(runtime.inspect(second)?.state, Value::Integer(42));
        assert!(runtime.send(first, test_message(0, 2, None)).is_err());
        runtime.kill(first)?;
        assert_eq!(runtime.inspect(first)?.status, Status::Stopped);
        runtime.restart(first)?;
        assert_eq!(runtime.inspect(first)?.incarnation, 2);
        assert_eq!(runtime.inspect(first)?.queued.len(), 1);
        Ok(())
    }

    #[test]
    fn execution_snapshot_reports_live_progress_and_is_stable_while_paused() -> ExperimentResult<()>
    {
        let mut runtime = Runtime::default();
        let implementation = publish(
            &mut runtime,
            "pub fn handle(ctx, state, sender, message) { loop {} }",
        )?;
        let actor = runtime.spawn(implementation, Value::Integer(0), capacity(2))?;
        runtime.send(actor, test_message(0, 1, None))?;
        runtime.tick(capacity(3));

        let running = runtime.inspect(actor)?;
        assert_eq!(running.status, Status::Running);
        let progress = running
            .execution
            .ok_or("running actor has no execution snapshot")?;
        assert!(progress.vm_steps > 0);
        assert!(running.reserved.is_some());

        runtime.pause(actor)?;
        for _ in 0..5 {
            runtime.tick(capacity(10));
        }
        let paused = runtime.inspect(actor)?;
        assert_eq!(paused.status, Status::Paused);
        assert_eq!(paused.execution, Some(progress));

        runtime.cancel(actor)?;
        let cancelled = runtime.inspect(actor)?;
        assert_eq!(cancelled.execution, None);
        assert_eq!(cancelled.queued, vec![test_message(0, 1, None)]);
        Ok(())
    }

    #[test]
    fn invalid_effect_does_not_commit_state_or_partial_sends() -> ExperimentResult<()> {
        let mut runtime = Runtime::default();
        let code = publish(
            &mut runtime,
            "pub fn handle(ctx, state, sender, message) { ctx.send(message.target, #{kind: 0, value: 1, target: ()}); ctx.send(\"invalid\", #{kind: 0, value: 2, target: ()}); 99 }",
        )?;
        let actor = runtime.spawn(code, Value::Integer(5), capacity(4))?;
        runtime.send(actor, test_message(0, 0, Some(actor)))?;
        for _ in 0..100 {
            runtime.tick(capacity(10));
        }
        let snapshot = runtime.inspect(actor)?;
        assert_eq!(snapshot.state, Value::Integer(5));
        assert!(matches!(snapshot.status, Status::Failed(_)));
        assert!(snapshot.queued.is_empty());
        assert!(snapshot.reserved.is_some());
        Ok(())
    }

    #[test]
    fn awaiting_retains_message_and_state_until_woken() -> ExperimentResult<()> {
        use std::future::Future;
        use std::pin::Pin;
        use std::sync::Mutex;
        use std::task::Poll;

        #[derive(Default)]
        struct Gate {
            ready: AtomicBool,
            dropped: AtomicBool,
            polls: std::sync::atomic::AtomicUsize,
            waker: Mutex<Option<Waker>>,
        }
        struct Wait(Arc<Gate>);
        impl Future for Wait {
            type Output = i64;
            fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<i64> {
                self.0.polls.fetch_add(1, Ordering::SeqCst);
                if self.0.ready.load(Ordering::SeqCst) {
                    Poll::Ready(7)
                } else {
                    *self.0.waker.lock().unwrap() = Some(cx.waker().clone());
                    Poll::Pending
                }
            }
        }
        impl Drop for Wait {
            fn drop(&mut self) {
                self.0.dropped.store(true, Ordering::SeqCst);
            }
        }

        let gate = Arc::new(Gate::default());
        let native_gate = Arc::clone(&gate);
        let mut module = rune::Module::new();
        module
            .function("wait", move || Wait(Arc::clone(&native_gate)))
            .build()?;
        let mut runtime = Runtime::default();
        let code = runtime.publish(
            CodeStore::default().compile_with_module(
                "pub async fn handle(ctx, state, sender, message) { if message.kind == 1 { let x = wait().await; state + x } else { state + message.value } }",
                "test-gate:v1",
                module,
            )?,
        )?;
        let actor = runtime.spawn(code, Value::Integer(10), capacity(2))?;
        runtime.send(actor, test_message(1, 0, None))?;
        runtime.send(actor, test_message(0, 3, None))?;
        for _ in 0..100 {
            runtime.tick(capacity(5));
        }
        let pending = runtime.inspect(actor)?;
        assert_eq!(pending.status, Status::Awaiting);
        assert_eq!(pending.state, Value::Integer(10));
        assert_eq!(pending.queued.len(), 1);
        assert_eq!(pending.reserved, Some(test_message(1, 0, None)));
        let polls = gate.polls.load(Ordering::SeqCst);
        for _ in 0..10 {
            runtime.tick(capacity(5));
        }
        assert_eq!(gate.polls.load(Ordering::SeqCst), polls);
        runtime.pause(actor)?;
        assert_eq!(runtime.inspect(actor)?.status, Status::Paused);
        gate.ready.store(true, Ordering::SeqCst);
        gate.waker
            .lock()
            .unwrap()
            .take()
            .expect("registered wake")
            .wake();
        for _ in 0..10 {
            runtime.tick(capacity(5));
        }
        assert_eq!(gate.polls.load(Ordering::SeqCst), polls);
        assert_eq!(runtime.inspect(actor)?.status, Status::Paused);
        runtime.resume(actor)?;
        for _ in 0..100 {
            runtime.tick(capacity(5));
        }
        let completed = runtime.inspect(actor)?;
        assert_eq!(completed.state, Value::Integer(20));
        assert_eq!(completed.status, Status::Idle);
        assert!(completed.reserved.is_none());
        assert!(completed.queued.is_empty());

        gate.ready.store(false, Ordering::SeqCst);
        gate.dropped.store(false, Ordering::SeqCst);
        runtime.send(actor, test_message(1, 0, None))?;
        for _ in 0..100 {
            runtime.tick(capacity(5));
        }
        assert_eq!(runtime.inspect(actor)?.status, Status::Awaiting);
        runtime.cancel(actor)?;
        assert!(gate.dropped.load(Ordering::SeqCst));
        let cancelled = runtime.inspect(actor)?;
        assert_eq!(cancelled.mailbox, pending.mailbox);
        assert_eq!(cancelled.state, Value::Integer(20));
        assert_eq!(cancelled.status, Status::Idle);
        assert_eq!(cancelled.queued, vec![test_message(1, 0, None)]);
        assert!(cancelled.reserved.is_none());
        Ok(())
    }

    #[test]
    fn pause_preserves_active_invocation_and_resume_continues_it() -> ExperimentResult<()> {
        let mut runtime = Runtime::default();
        let implementation = publish(
            &mut runtime,
            "pub fn handle(ctx, state, sender, message) { state + message.value }",
        )?;
        let actor = runtime.spawn(implementation, Value::Integer(10), capacity(2))?;
        runtime.send(actor, test_message(0, 7, None))?;
        runtime.tick(capacity(1));
        let before_pause = runtime.inspect(actor)?;
        assert_eq!(before_pause.status, Status::Running);
        assert_eq!(before_pause.state, Value::Integer(10));
        assert!(before_pause.reserved.is_some());

        runtime.pause(actor)?;
        assert!(runtime.pause(actor).is_err());
        for _ in 0..10 {
            runtime.tick(capacity(10));
        }
        let paused = runtime.inspect(actor)?;
        assert_eq!(paused.status, Status::Paused);
        assert_eq!(paused.state, before_pause.state);
        assert_eq!(paused.reserved, before_pause.reserved);

        runtime.resume(actor)?;
        assert!(runtime.resume(actor).is_err());
        for _ in 0..100 {
            runtime.tick(capacity(10));
        }
        let completed = runtime.inspect(actor)?;
        assert_eq!(completed.status, Status::Idle);
        assert_eq!(completed.state, Value::Integer(17));
        assert!(completed.reserved.is_none());
        let events = runtime.drain_events();
        assert!(
            events
                .iter()
                .any(|event| matches!(event, Event::Paused(id) if *id == actor))
        );
        assert!(
            events
                .iter()
                .any(|event| matches!(event, Event::Resumed(id) if *id == actor))
        );
        Ok(())
    }

    #[test]
    fn mailbox_capacity_and_runtime_failures_are_explicit() -> ExperimentResult<()> {
        let mut runtime = Runtime::default();
        let code = publish(
            &mut runtime,
            "pub fn handle(ctx, state, sender, message) { state / message.value }",
        )?;
        let actor = runtime.spawn(code, Value::Integer(5), capacity(1))?;
        runtime.send(actor, test_message(0, 0, None))?;
        assert!(runtime.send(actor, test_message(0, 1, None)).is_err());
        assert!(runtime.inspect(ActorRef::new()).is_err());
        for _ in 0..100 {
            runtime.tick(capacity(5));
        }
        assert!(matches!(runtime.inspect(actor)?.status, Status::Failed(_)));
        assert_eq!(runtime.inspect(actor)?.state, Value::Integer(5));
        assert!(
            runtime
                .drain_events()
                .iter()
                .any(|e| matches!(e, Event::Failed(id, _) if *id == actor))
        );
        Ok(())
    }

    #[test]
    fn aggregate_self_sends_respect_capacity_without_partial_commit() -> ExperimentResult<()> {
        let mut runtime = Runtime::default();
        let code = publish(
            &mut runtime,
            "pub fn handle(ctx, state, sender, message) { ctx.send(message.target, #{kind: 0, value: 1, target: ()}); ctx.send(message.target, #{kind: 0, value: 2, target: ()}); 99 }",
        )?;
        let actor = runtime.spawn(code, Value::Integer(5), capacity(1))?;
        runtime.send(actor, test_message(0, 0, Some(actor)))?;
        for _ in 0..100 {
            runtime.tick(capacity(5));
        }
        let snapshot = runtime.inspect(actor)?;
        assert!(matches!(snapshot.status, Status::Failed(_)));
        assert_eq!(snapshot.state, Value::Integer(5));
        assert!(snapshot.queued.is_empty());
        assert!(snapshot.reserved.is_some());
        Ok(())
    }

    #[test]
    fn invalid_transition_shape_fails_without_committing() -> ExperimentResult<()> {
        let mut runtime = Runtime::default();
        let code = publish(
            &mut runtime,
            "pub fn handle(ctx, state, sender, message) { Some(99) }",
        )?;
        let actor = runtime.spawn(code, Value::Integer(5), capacity(1))?;
        runtime.send(actor, test_message(0, 1, None))?;
        for _ in 0..100 {
            runtime.tick(capacity(5));
        }
        let snapshot = runtime.inspect(actor)?;
        assert!(matches!(snapshot.status, Status::Failed(_)));
        assert_eq!(snapshot.state, Value::Integer(5));
        assert!(snapshot.reserved.is_some());
        Ok(())
    }

    #[test]
    fn behavior_installation_waits_for_handler_boundary() -> ExperimentResult<()> {
        let mut code = CodeStore::default();
        let first =
            code.compile("pub fn handle(ctx, state, sender, message) { state + message.value }")?;
        let second = code
            .compile("pub fn handle(ctx, state, sender, message) { state + message.value * 10 }")?;
        let first_id = first.implementation_id;
        let second_id = second.implementation_id;
        let mut runtime = Runtime::default();
        runtime.publish(first)?;
        runtime.publish(second)?;
        let actor = runtime.spawn(first_id, Value::Integer(0), capacity(2))?;
        runtime.send(actor, test_message(0, 1, None))?;
        runtime.tick(capacity(1));
        let active = runtime.inspect(actor)?;
        assert_eq!(active.implementation, first_id);
        assert!(runtime.install(actor, second_id).is_err());
        for _ in 0..100 {
            runtime.tick(capacity(10));
        }
        assert_eq!(runtime.inspect(actor)?.state, Value::Integer(1));
        runtime.send(actor, test_message(0, 2, None))?;
        runtime.install(actor, second_id)?;
        assert_eq!(runtime.inspect(actor)?.implementation, second_id);
        for _ in 0..100 {
            runtime.tick(capacity(10));
        }
        let updated = runtime.inspect(actor)?;
        assert_eq!(updated.state, Value::Integer(21));
        assert_eq!(updated.implementation, second_id);
        assert!(
            runtime
                .drain_events()
                .iter()
                .any(|event| matches!(event, Event::BehaviorInstalled(id, version) if *id == actor && *version == second_id))
        );
        Ok(())
    }

    #[test]
    fn behavior_installation_migrates_state_atomically() -> ExperimentResult<()> {
        let mut runtime = Runtime::default();
        let old_implementation = publish(
            &mut runtime,
            "pub fn handle(ctx, state, sender, message) { state + message.value }",
        )?;
        let new_implementation = publish(
            &mut runtime,
            "pub fn handle(ctx, state, sender, message) { state * message.value }",
        )?;
        let actor = runtime.spawn(old_implementation, Value::Integer(7), capacity(2))?;

        runtime.install_with_migration(actor, new_implementation, |state| {
            let Value::Integer(state) = state else {
                return Err("expected integer".into());
            };
            Ok(Value::Integer(
                state.checked_mul(10).ok_or("state migration overflow")?,
            ))
        })?;

        let migrated = runtime.inspect(actor)?;
        assert_eq!(migrated.implementation, new_implementation);
        assert_eq!(migrated.state, Value::Integer(70));
        assert!(
            runtime
                .drain_events()
                .iter()
                .any(|event| matches!(event, Event::StateMigrated(id, Value::Integer(7), Value::Integer(70)) if *id == actor))
        );
        Ok(())
    }

    #[test]
    fn cancelled_input_retries_first_with_new_behavior() -> ExperimentResult<()> {
        let mut runtime = Runtime::default();
        let old_implementation = publish(
            &mut runtime,
            "pub fn handle(ctx, state, sender, message) { state + message.value }",
        )?;
        let new_implementation = publish(
            &mut runtime,
            "pub fn handle(ctx, state, sender, message) { state * 10 + message.value }",
        )?;
        let actor = runtime.spawn(old_implementation, Value::Integer(0), capacity(4))?;
        runtime.send(actor, test_message(0, 1, None))?;
        runtime.send(actor, test_message(0, 2, None))?;
        runtime.tick(capacity(1));

        assert_eq!(runtime.inspect(actor)?.status, Status::Running);
        runtime.cancel(actor)?;
        let cancelled = runtime.inspect(actor)?;
        assert_eq!(cancelled.status, Status::Idle);
        assert_eq!(cancelled.state, Value::Integer(0));
        assert_eq!(
            cancelled.queued,
            vec![test_message(0, 1, None), test_message(0, 2, None)]
        );
        assert!(cancelled.reserved.is_none());

        runtime.install(actor, new_implementation)?;
        runtime.send(actor, test_message(0, 3, None))?;
        for _ in 0..100 {
            runtime.tick(capacity(10));
        }
        let completed = runtime.inspect(actor)?;
        assert_eq!(completed.state, Value::Integer(123));
        assert_eq!(completed.implementation, new_implementation);
        assert!(
            runtime
                .drain_events()
                .iter()
                .any(|event| matches!(event, Event::InvocationCancelled(id) if *id == actor))
        );
        Ok(())
    }

    #[test]
    fn failed_state_migration_preserves_actor_state_and_implementation() -> ExperimentResult<()> {
        let mut runtime = Runtime::default();
        let old_implementation = publish(
            &mut runtime,
            "pub fn handle(ctx, state, sender, message) { state + message.value }",
        )?;
        let new_implementation = publish(
            &mut runtime,
            "pub fn handle(ctx, state, sender, message) { state * message.value }",
        )?;
        let actor = runtime.spawn(old_implementation, Value::Integer(7), capacity(2))?;

        assert!(
            runtime
                .install_with_migration(actor, new_implementation, |_| {
                    Err("unsupported state version".into())
                })
                .is_err()
        );

        let unchanged = runtime.inspect(actor)?;
        assert_eq!(unchanged.implementation, old_implementation);
        assert_eq!(unchanged.state, Value::Integer(7));
        assert!(
            !runtime
                .drain_events()
                .iter()
                .any(|event| matches!(event, Event::BehaviorInstalled(id, _) if *id == actor))
        );
        Ok(())
    }

    #[test]
    fn migration_is_not_run_outside_an_idle_handler_boundary() -> ExperimentResult<()> {
        use std::cell::Cell;

        let mut runtime = Runtime::default();
        let old_implementation = publish(
            &mut runtime,
            "pub fn handle(ctx, state, sender, message) { state + message.value }",
        )?;
        let new_implementation = publish(
            &mut runtime,
            "pub fn handle(ctx, state, sender, message) { state * message.value }",
        )?;
        let actor = runtime.spawn(old_implementation, Value::Integer(7), capacity(2))?;
        runtime.send(actor, test_message(0, 1, None))?;
        runtime.tick(capacity(1));
        let called = Cell::new(false);

        assert!(
            runtime
                .install_with_migration(actor, new_implementation, |state| {
                    called.set(true);
                    let Value::Integer(state) = state else {
                        return Err("expected integer".into());
                    };
                    Ok(Value::Integer(state + 100))
                })
                .is_err()
        );
        assert!(!called.get());
        assert_eq!(runtime.inspect(actor)?.implementation, old_implementation);
        Ok(())
    }

    #[test]
    fn restart_retries_failed_reservation_from_last_committed_state() -> ExperimentResult<()> {
        use std::sync::atomic::AtomicBool;

        let fail_once = Arc::new(AtomicBool::new(true));
        let native_fail_once = Arc::clone(&fail_once);
        let mut module = rune::Module::new();
        module
            .function("fail_once", move || {
                native_fail_once.swap(false, Ordering::SeqCst)
            })
            .build()?;
        let mut runtime = Runtime::default();
        let code = runtime.publish(
            CodeStore::default().compile_with_module(
                "fn check() { if fail_once() { 1 / 0 } else { 0 } } pub fn handle(ctx, state, sender, message) { let _ = check(); state + message.value }",
                "test-fail-once:v1",
                module,
            )?,
        )?;
        let actor = runtime.spawn(code, Value::Integer(10), capacity(3))?;
        let first = test_message(1, 2, None);
        let second = test_message(1, 3, None);
        runtime.send(actor, first.clone())?;
        runtime.send(actor, second.clone())?;
        for _ in 0..100 {
            runtime.tick(capacity(5));
        }

        let failed = runtime.inspect(actor)?;
        assert!(matches!(failed.status, Status::Failed(_)));
        assert_eq!(failed.state, Value::Integer(10));
        assert_eq!(failed.reserved, Some(first.clone()));
        assert_eq!(failed.queued, vec![second.clone()]);
        assert!(runtime.send(actor, test_message(1, 4, None)).is_err());
        assert!(
            runtime
                .drain_events()
                .iter()
                .any(|event| matches!(event, Event::Failed(id, _) if *id == actor))
        );

        runtime.restart(actor)?;
        let restarted = runtime.inspect(actor)?;
        assert_eq!(restarted.actor_ref, failed.actor_ref);
        assert_eq!(restarted.mailbox, failed.mailbox);
        assert_eq!(restarted.implementation, failed.implementation);
        assert_eq!(restarted.incarnation, failed.incarnation + 1);
        assert_eq!(restarted.state, Value::Integer(10));
        assert_eq!(restarted.queued, vec![first.clone(), second.clone()]);
        assert_eq!(restarted.reserved, None);

        for _ in 0..100 {
            runtime.tick(capacity(5));
        }
        let completed = runtime.inspect(actor)?;
        assert_eq!(completed.state, Value::Integer(15), "{completed:?}");
        assert_eq!(completed.status, Status::Idle);
        assert_eq!(completed.queued, Vec::<Message>::new());
        assert_eq!(completed.reserved, None);
        Ok(())
    }

    #[test]
    fn stop_and_restart_preserve_reserved_and_queued_fifo_order() -> ExperimentResult<()> {
        let mut runtime = Runtime::default();
        let code = publish(
            &mut runtime,
            "pub fn handle(ctx, state, sender, message) { state * 10 + message.value }",
        )?;
        let actor = runtime.spawn(code, Value::Integer(0), capacity(3))?;
        let first = test_message(0, 1, None);
        let second = test_message(0, 2, None);
        let third = test_message(0, 3, None);
        runtime.send(actor, first.clone())?;
        runtime.send(actor, second.clone())?;
        runtime.send(actor, third.clone())?;

        runtime.tick(capacity(1));
        let running = runtime.inspect(actor)?;
        assert_eq!(running.status, Status::Running);
        assert_eq!(running.state, Value::Integer(0));
        assert_eq!(running.reserved, Some(first.clone()));
        assert_eq!(running.queued, vec![second.clone(), third.clone()]);

        runtime.kill(actor)?;
        let stopped = runtime.inspect(actor)?;
        assert_eq!(stopped.status, Status::Stopped);
        assert_eq!(stopped.state, Value::Integer(0));
        assert_eq!(stopped.reserved, Some(first.clone()));
        assert_eq!(stopped.queued, vec![second.clone(), third.clone()]);
        assert!(runtime.send(actor, test_message(0, 4, None)).is_err());

        runtime.restart(actor)?;
        let restarted = runtime.inspect(actor)?;
        assert_eq!(restarted.state, Value::Integer(0));
        assert_eq!(restarted.reserved, None);
        assert_eq!(
            restarted.queued,
            vec![first.clone(), second.clone(), third.clone()]
        );
        for _ in 0..100 {
            runtime.tick(capacity(5));
        }
        let completed = runtime.inspect(actor)?;
        assert_eq!(completed.state, Value::Integer(123));
        assert_eq!(completed.status, Status::Idle);
        assert_eq!(completed.queued, Vec::<Message>::new());
        assert!(completed.reserved.is_none());
        Ok(())
    }
}