//! Trusted, in-process application services for a running Runa image.

use std::collections::BTreeMap;
use std::num::NonZeroUsize;
use std::sync::Arc;

use runa_actor::{ActorRef, ActorSnapshot, Event, Message, SystemMessage};
use runa_code::{BehaviorSnapshot, CodeStore, SourceCommit, SourceFile};
use runa_core::{ImplementationId, Value};
use runa_runtime::{Runtime, RuntimeResult};

pub struct ImageApi {
    code: CodeStore,
    runtime: Runtime,
    actor_paths: BTreeMap<ActorRef, String>,
    pending_paths: BTreeMap<ActorRef, (String, ImplementationId)>,
    temporary: BTreeMap<ActorRef, SourceFile>,
    events: Vec<Event>,
}

pub struct ExecutionInspection {
    pub actor_ref: ActorRef,
    pub incarnation: u64,
    pub implementation_id: ImplementationId,
    pub source_id: runa_core::SourceId,
    pub source: String,
    pub behavior_path: Option<String>,
    pub execution: Option<runa_core::ExecutionInspection>,
    pub debugger: runa_core::DebuggerSnapshot,
}

impl Default for ImageApi {
    fn default() -> Self {
        Self::new()
    }
}

impl ImageApi {
    pub fn new() -> Self {
        Self {
            code: CodeStore::default(),
            runtime: Runtime::default(),
            actor_paths: BTreeMap::new(),
            pending_paths: BTreeMap::new(),
            temporary: BTreeMap::new(),
            events: Vec::new(),
        }
    }

    pub fn publish_behavior(
        &mut self,
        name: &str,
        expected_revision: Option<u64>,
        source: &str,
    ) -> RuntimeResult<BehaviorSnapshot> {
        let artifact = self.code.compile(source)?;
        self.runtime.publish(Arc::clone(&artifact))?;
        let behavior = self
            .code
            .publish_behavior(name, expected_revision, source)?;
        if runa_code::validate_source_path(name).is_ok() {
            let expected_source_id = self.code.source(name).map(|file| file.source_id);
            self.code.write_source(name, expected_source_id, source)?;
        }
        Ok(behavior)
    }

    pub fn rollback_behavior(
        &mut self,
        name: &str,
        expected_revision: u64,
        target_revision: u64,
    ) -> RuntimeResult<BehaviorSnapshot> {
        self.code
            .rollback_behavior(name, expected_revision, target_revision)
    }

    pub fn behavior(&self, name: &str) -> Option<BehaviorSnapshot> {
        self.code.behavior(name)
    }

    pub fn behaviors(&self) -> Vec<BehaviorSnapshot> {
        self.code.behaviors()
    }

    pub fn behavior_source(&self, name: &str, revision: u64) -> RuntimeResult<&str> {
        let behavior = self.code.behavior(name).ok_or("unknown behavior")?;
        let version = behavior
            .history
            .iter()
            .find(|version| version.revision == revision)
            .ok_or("unknown behavior revision")?;
        self.code
            .artifact(version.implementation_id)
            .map(|artifact| artifact.original_source())
            .ok_or_else(|| "behavior artifact is unavailable".into())
    }

    pub fn write_source(
        &mut self,
        path: &str,
        expected_source_id: Option<runa_core::SourceId>,
        source: &str,
    ) -> RuntimeResult<SourceFile> {
        self.code.write_source(path, expected_source_id, source)
    }

    pub fn sources(&self) -> Vec<SourceFile> {
        self.code.sources()
    }

    pub fn check_source(&self, source: &str) -> RuntimeResult<Vec<runa_core::SourceDiagnostic>> {
        self.code.check_source(source)
    }

    pub fn source(&self, path: &str) -> Option<&SourceFile> {
        self.code.source(path)
    }

    pub fn commit_source(&mut self, path: &str) -> RuntimeResult<SourceCommit> {
        self.code.commit_source(path)
    }

    pub fn source_history(&self, path: &str) -> Option<&[SourceCommit]> {
        self.code.source_history(path)
    }

    pub fn source_commit(&self, path: &str, revision: &str) -> Option<&SourceCommit> {
        self.code.source_commit(path, revision)
    }

    pub fn source_commit_source(&self, path: &str, revision: &str) -> Option<&str> {
        self.code.source_commit_source(path, revision)
    }

    pub fn rollback_source(&mut self, path: &str, revision: &str) -> RuntimeResult<SourceFile> {
        self.code.rollback_source(path, revision)
    }

    pub fn source_users(&self, path: &str) -> Vec<ActorRef> {
        self.runtime
            .actors()
            .into_iter()
            .filter(|actor| {
                self.actor_paths
                    .get(&actor.actor_ref)
                    .is_some_and(|p| p == path)
                    || self
                        .pending_paths
                        .get(&actor.actor_ref)
                        .is_some_and(|(p, _)| p == path)
            })
            .map(|actor| actor.actor_ref)
            .collect()
    }

    pub fn rename_source(
        &mut self,
        path: &str,
        target: &str,
        expected: runa_core::SourceId,
        update_actors: bool,
    ) -> RuntimeResult<SourceFile> {
        if !update_actors && !self.source_users(path).is_empty() {
            return Err("behavior is in use; confirm updating actors before renaming".into());
        }
        let file = self.code.rename_source(path, target, expected)?;
        for value in self.actor_paths.values_mut() {
            if value == path {
                *value = target.to_owned();
            }
        }
        for (value, _) in self.pending_paths.values_mut() {
            if value == path {
                *value = target.to_owned();
            }
        }
        for file in self.temporary.values_mut() {
            if file.path == path {
                file.path = target.to_owned();
            }
        }
        Ok(file)
    }

    pub fn delete_source(
        &mut self,
        path: &str,
        expected: runa_core::SourceId,
    ) -> RuntimeResult<()> {
        if self.source_users(path).iter().any(|id| {
            self.runtime
                .inspect(*id)
                .is_ok_and(|actor| !matches!(actor.status, runa_actor::Status::Failed(_)))
        }) {
            return Err("behavior is used by running actors; stop the actors first".into());
        }
        self.code.delete_source(path, expected)
    }

    pub fn actor_path(&self, actor: ActorRef) -> Option<&str> {
        self.actor_paths.get(&actor).map(String::as_str)
    }
    pub fn pending_actor_path(&self, actor: ActorRef) -> Option<&str> {
        self.pending_paths
            .get(&actor)
            .map(|(path, _)| path.as_str())
    }

    pub fn has_temporary_behavior(&self, actor: ActorRef) -> bool {
        self.temporary.contains_key(&actor)
    }

    pub fn actor_behavior(&self, actor: ActorRef) -> RuntimeResult<SourceFile> {
        let snapshot = self.runtime.inspect(actor)?;
        if let Some(file) = self.temporary.get(&actor) {
            return Ok(file.clone());
        }
        let artifact = self
            .code
            .artifact(snapshot.implementation)
            .ok_or("behavior artifact is unavailable")?;
        Ok(SourceFile {
            path: self
                .actor_path(actor)
                .ok_or("actor has no source path")?
                .to_owned(),
            source_id: artifact.source_id,
            source: artifact.original_source().to_owned(),
        })
    }

    pub fn write_actor_behavior(
        &mut self,
        actor: ActorRef,
        expected: String,
        source: &str,
    ) -> RuntimeResult<SourceFile> {
        let previous = self.actor_behavior(actor)?;
        if previous.source_id.to_string() != expected {
            return Err("source identity conflict: actor behavior changed".into());
        }
        let artifact = self.code.compile(source)?;
        self.runtime.publish(Arc::clone(&artifact))?;
        self.runtime
            .set_actor_behavior(actor, artifact.implementation_id)?;
        self.pending_paths.remove(&actor);
        let file = SourceFile {
            path: previous.path,
            source_id: artifact.source_id,
            source: source.to_owned(),
        };
        self.temporary.insert(actor, file.clone());
        self.collect_runtime_events();
        Ok(file)
    }

    pub fn update_behavior_source(
        &mut self,
        actor: ActorRef,
        expected: runa_core::SourceId,
        temporary_id: &str,
    ) -> RuntimeResult<SourceFile> {
        self.runtime.inspect(actor)?;
        let file = self
            .temporary
            .get(&actor)
            .ok_or("actor has no temporary behavior")?;
        if file.source_id.to_string() != temporary_id {
            return Err("source identity conflict: temporary behavior changed".into());
        }
        let path = self
            .actor_paths
            .get(&actor)
            .ok_or("actor has no source path")?;
        self.code.write_source(path, Some(expected), &file.source)
    }

    pub fn spawn(
        &mut self,
        path: &str,
        state: Value,
        capacity: NonZeroUsize,
    ) -> RuntimeResult<ActorRef> {
        let source = self
            .code
            .source(path)
            .ok_or("unknown source path")?
            .source
            .clone();
        let artifact = self.code.compile(&source)?;
        self.runtime.publish(Arc::clone(&artifact))?;
        let actor = self
            .runtime
            .spawn(artifact.implementation_id, state, capacity)?;
        self.actor_paths.insert(actor, path.to_owned());
        Ok(actor)
    }

    pub fn reload(&mut self, actor: ActorRef) -> RuntimeResult<ImplementationId> {
        let path = self
            .actor_paths
            .get(&actor)
            .cloned()
            .ok_or("actor has no source path")?;
        let result = self.schedule_source_install(actor, &path)?;
        self.temporary.remove(&actor);
        Ok(result)
    }

    pub fn r#become(&mut self, actor: ActorRef, path: &str) -> RuntimeResult<ImplementationId> {
        self.schedule_source_install(actor, path)
    }

    fn schedule_source_install(
        &mut self,
        actor: ActorRef,
        path: &str,
    ) -> RuntimeResult<ImplementationId> {
        let source = self
            .code
            .source(path)
            .ok_or("unknown source path")?
            .source
            .clone();
        let artifact = self.code.compile(&source)?;
        self.runtime.publish(Arc::clone(&artifact))?;
        self.runtime
            .set_actor_behavior(actor, artifact.implementation_id)?;
        self.pending_paths
            .insert(actor, (path.to_owned(), artifact.implementation_id));
        self.temporary.remove(&actor);
        self.collect_runtime_events();
        Ok(artifact.implementation_id)
    }

    pub fn send(&mut self, actor: ActorRef, message: Message) -> RuntimeResult<()> {
        self.runtime.send(actor, message)
    }

    pub fn tick(&mut self, ticks: usize, budget: NonZeroUsize) {
        for _ in 0..ticks {
            self.runtime.tick(budget);
            self.collect_runtime_events();
        }
    }

    pub fn inspect(&self, actor: ActorRef) -> RuntimeResult<ActorSnapshot> {
        self.runtime.inspect(actor)
    }

    pub fn inspect_execution(&self, actor: ActorRef) -> RuntimeResult<ExecutionInspection> {
        let snapshot = self.runtime.inspect(actor)?;
        let artifact = self
            .code
            .artifact(snapshot.implementation)
            .ok_or("behavior artifact is unavailable")?;
        Ok(ExecutionInspection {
            actor_ref: actor,
            incarnation: snapshot.incarnation,
            implementation_id: snapshot.implementation,
            source_id: artifact.source_id,
            source: artifact.original_source().to_owned(),
            behavior_path: self.actor_path(actor).map(str::to_owned),
            execution: self.runtime.inspect_execution(actor)?,
            debugger: self.runtime.debugger(actor)?,
        })
    }

    pub fn set_breakpoints(
        &mut self,
        actor: ActorRef,
        expected: ImplementationId,
        lines: &[usize],
    ) -> RuntimeResult<Vec<runa_core::SourceBreakpoint>> {
        self.runtime.set_breakpoints(actor, expected, lines)
    }

    pub fn step_source(
        &mut self,
        actor: ActorRef,
        expected: ImplementationId,
        budget: NonZeroUsize,
    ) -> RuntimeResult<runa_core::DebugStopReason> {
        let result = self.runtime.step_source(actor, expected, budget);
        self.collect_runtime_events();
        result
    }

    pub fn edit_actor(
        &mut self,
        actor: ActorRef,
        state: Option<Value>,
        messages: Option<(Option<Value>, Vec<Value>)>,
    ) -> RuntimeResult<()> {
        self.runtime.edit_actor(actor, state, messages)
    }

    pub fn set_log_level(
        &mut self,
        actor: ActorRef,
        level: runa_core::LogLevel,
    ) -> RuntimeResult<()> {
        self.runtime.set_log_level(actor, level)
    }

    pub fn log_level(&self, actor: ActorRef) -> RuntimeResult<runa_core::LogLevel> {
        self.runtime.log_level(actor)
    }

    pub fn logs(&self, actors: &[ActorRef], after: u64) -> RuntimeResult<Vec<runa_core::LogEntry>> {
        self.runtime.logs(actors, after)
    }

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

    pub fn actors(&self) -> Vec<ActorSnapshot> {
        self.runtime.actors()
    }

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

    fn collect_runtime_events(&mut self) {
        for event in self.runtime.drain_events() {
            if let Event::Stopped(actor) = &event {
                self.actor_paths.remove(actor);
                self.pending_paths.remove(actor);
                self.temporary.remove(actor);
            }
            if let Event::BehaviorInstalled(actor, implementation) = &event
                && self
                    .pending_paths
                    .get(actor)
                    .is_some_and(|(_, pending)| pending == implementation)
            {
                let (path, _) = self
                    .pending_paths
                    .remove(actor)
                    .expect("pending path exists");
                self.actor_paths.insert(*actor, path);
            }
            self.events.push(event);
        }
    }

    pub fn system(&mut self, actor: ActorRef, message: SystemMessage) -> RuntimeResult<()> {
        self.collect_runtime_events();
        self.runtime.system(actor, message)?;
        if matches!(message, SystemMessage::Kill | SystemMessage::Restart) {
            self.pending_paths.remove(&actor);
        }
        if message == SystemMessage::Restart
            && let Some(file) = self.temporary.get(&actor)
        {
            let artifact = self.code.compile(&file.source)?;
            self.runtime
                .set_actor_behavior(actor, artifact.implementation_id)?;
        }
        self.collect_runtime_events();
        Ok(())
    }

    pub fn stop(&mut self, actor: ActorRef) -> RuntimeResult<()> {
        self.system(actor, SystemMessage::Stop)
    }

    pub fn restart(&mut self, actor: ActorRef) -> RuntimeResult<()> {
        self.system(actor, SystemMessage::Restart)
    }

    pub fn pause(&mut self, actor: ActorRef) -> RuntimeResult<()> {
        self.system(actor, SystemMessage::Pause)
    }

    pub fn resume(&mut self, actor: ActorRef) -> RuntimeResult<()> {
        self.system(actor, SystemMessage::Resume)
    }

    pub fn cancel(&mut self, actor: ActorRef) -> RuntimeResult<()> {
        self.runtime.cancel(actor)
    }

    pub fn install(
        &mut self,
        actor: ActorRef,
        behavior_name: &str,
    ) -> RuntimeResult<ImplementationId> {
        self.install_with_migration(actor, behavior_name, Ok)
    }

    pub fn install_with_migration(
        &mut self,
        actor: ActorRef,
        behavior_name: &str,
        migrate: impl FnOnce(Value) -> RuntimeResult<Value>,
    ) -> RuntimeResult<ImplementationId> {
        let behavior = self
            .code
            .behavior(behavior_name)
            .ok_or("unknown behavior")?;
        self.runtime
            .install_with_migration(actor, behavior.current, migrate)?;
        Ok(behavior.current)
    }
}

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

    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),
            ),
        ]))
    }

    fn capacity(n: usize) -> NonZeroUsize {
        NonZeroUsize::new(n).expect("capacity must be positive")
    }

    #[test]
    fn image_api_publishes_spawns_inspects_and_exposes_events() -> RuntimeResult<()> {
        let mut image = ImageApi::new();
        let behavior = image.publish_behavior(
            "counter",
            None,
            "pub fn handle(ctx, state, sender, message) { state + message.value }",
        )?;
        assert_eq!(
            image.behavior_source("counter", behavior.revision)?,
            "pub fn handle(ctx, state, sender, message) { state + message.value }"
        );
        let actor = image.spawn("counter", Value::Integer(5), capacity(2))?;
        image.send(actor, test_message(0, 7, None))?;
        image.tick(100, capacity(10));

        assert_eq!(image.inspect(actor)?.state, Value::Integer(12));
        assert_eq!(image.inspect(actor)?.implementation, behavior.current);
        assert!(
            image
                .events()
                .iter()
                .any(|event| matches!(event, Event::Committed(id) if id == &actor))
        );
        Ok(())
    }

    #[test]
    fn image_api_rejects_stale_behavior_updates_without_changing_current() -> RuntimeResult<()> {
        let mut image = ImageApi::new();
        let first = image.publish_behavior(
            "counter",
            None,
            "pub fn handle(ctx, state, sender, message) { state + message.value }",
        )?;
        assert!(
            image
                .publish_behavior(
                    "counter",
                    None,
                    "pub fn handle(ctx, state, sender, message) { state - message.value }"
                )
                .is_err()
        );
        assert_eq!(image.behavior("counter").unwrap().current, first.current);
        Ok(())
    }

    #[test]
    fn image_api_installs_behavior_with_atomic_state_migration() -> RuntimeResult<()> {
        let mut image = ImageApi::new();
        let first = image.publish_behavior(
            "counter",
            None,
            "pub fn handle(ctx, state, sender, message) { state + message.value }",
        )?;
        image.publish_behavior(
            "counter",
            Some(first.revision),
            "pub fn handle(ctx, state, sender, message) { state * message.value }",
        )?;
        let actor = image.spawn("counter", Value::Integer(5), capacity(2))?;
        let implementation = image.install_with_migration(actor, "counter", |state| {
            let Value::Integer(state) = state else {
                return Err("expected integer".into());
            };
            Ok(Value::Integer(
                state.checked_mul(10).ok_or("migration overflow")?,
            ))
        })?;

        assert_eq!(image.inspect(actor)?.state, Value::Integer(50));
        assert_eq!(image.inspect(actor)?.implementation, implementation);
        Ok(())
    }

    #[test]
    fn spawn_compiles_current_uncommitted_source() -> RuntimeResult<()> {
        let mut image = ImageApi::new();
        let file = image.write_source(
            "actors/counter",
            None,
            "pub fn handle(ctx, state, sender, message) { state + message.value }",
        )?;
        let actor = image.spawn("actors/counter", Value::Integer(3), capacity(2))?;
        assert_eq!(
            image.inspect(actor)?.implementation,
            image.code.compile(&file.source)?.implementation_id
        );
        Ok(())
    }

    #[test]
    fn execution_inspection_keeps_exact_loaded_source_across_edits_pause_and_reload()
    -> RuntimeResult<()> {
        let mut image = ImageApi::new();
        let original = "pub fn handle(ctx, state, sender, message) {\n    nested()\n}\nfn nested() { loop {} }";
        let file = image.write_source("actors/stack", None, original)?;
        let actor = image.spawn(&file.path, Value::Null, capacity(2))?;
        assert!(image.inspect_execution(actor)?.execution.is_none());
        image.send(actor, Value::Null)?;
        image.tick(1, capacity(10));
        let before = image.inspect_execution(actor)?;
        assert_eq!(before.source, original);
        assert_eq!(before.source_id, file.source_id);
        assert_eq!(before.behavior_path.as_deref(), Some("actors/stack"));
        let frames = before.execution.as_ref().unwrap();
        assert_eq!(frames.frames[0].function.as_deref(), Some("nested"));
        assert_eq!(frames.frames[0].range.unwrap().start.line, 3);
        let updated = image.write_source(
            &file.path,
            Some(file.source_id),
            "pub fn handle(ctx, state, sender, message) { 1 }",
        )?;
        image.system(actor, SystemMessage::Pause)?;
        image.tick(1, capacity(1));
        let paused = image.inspect_execution(actor)?.execution.unwrap();
        image.tick(2, capacity(100));
        assert_eq!(image.inspect_execution(actor)?.execution, Some(paused));
        image.reload(actor)?;
        let pending = image.inspect_execution(actor)?;
        assert_eq!(pending.implementation_id, before.implementation_id);
        assert_eq!(pending.source, original);
        image.system(actor, SystemMessage::Restart)?;
        assert!(image.inspect_execution(actor)?.execution.is_none());
        assert_eq!(image.inspect_execution(actor)?.source, original);
        image.reload(actor)?;
        image.tick(2, capacity(100));
        let after = image.inspect_execution(actor)?;
        assert_eq!(after.source, updated.source);
        assert_eq!(after.source_id, updated.source_id);
        assert!(after.execution.is_none());
        assert_eq!(after.incarnation, before.incarnation + 1);
        image.system(actor, SystemMessage::Kill)?;
        assert!(image.inspect_execution(actor).is_err());
        Ok(())
    }

    #[test]
    fn execution_inspection_distinguishes_installed_and_pending_temporary_source()
    -> RuntimeResult<()> {
        let mut image = ImageApi::new();
        let original = image.write_source(
            "actors/temp_stack",
            None,
            "pub fn handle(ctx, state, sender, message) { state }",
        )?;
        let actor = image.spawn(&original.path, Value::Null, capacity(2))?;
        let temporary = image.write_actor_behavior(
            actor,
            original.source_id.to_string(),
            "pub fn handle(ctx, state, sender, message) { loop {} }",
        )?;
        assert_eq!(
            image.inspect_execution(actor)?.source_id,
            original.source_id
        );
        image.send(actor, Value::Null)?;
        image.tick(1, capacity(10));
        let running = image.inspect_execution(actor)?;
        assert_eq!(running.source_id, temporary.source_id);
        assert_eq!(running.source, temporary.source);
        assert!(running.execution.is_some());
        let replacement = image.write_actor_behavior(
            actor,
            temporary.source_id.to_string(),
            "pub fn handle(ctx, state, sender, message) { 42 }",
        )?;
        let pending = image.inspect_execution(actor)?;
        assert_eq!(pending.source_id, temporary.source_id);
        assert_ne!(pending.source_id, replacement.source_id);
        assert_eq!(pending.source, running.source);
        Ok(())
    }

    #[test]
    fn execution_inspection_maps_shutdown_and_has_no_frames_after_failure() -> RuntimeResult<()> {
        let mut image = ImageApi::new();
        let file = image.write_source("actors/shutdown_stack", None, "pub fn handle(ctx, state, sender, message) { state }\npub fn on_shutdown(state) { loop {} }")?;
        let actor = image.spawn(&file.path, Value::Null, capacity(2))?;
        image.system(actor, SystemMessage::Stop)?;
        image.tick(1, capacity(10));
        let shutdown = image.inspect_execution(actor)?.execution.unwrap();
        assert_eq!(shutdown.frames[0].function.as_deref(), Some("on_shutdown"));
        assert_eq!(shutdown.frames[0].range.unwrap().start.line, 1);
        let file = image.write_source(
            "actors/failing_stack",
            None,
            "pub fn handle(ctx, state, sender, message) { 1 / 0 }",
        )?;
        let failed = image.spawn(&file.path, Value::Null, capacity(2))?;
        image.send(failed, Value::Null)?;
        image.tick(1, capacity(100));
        let failed = image.inspect_execution(failed)?;
        assert!(failed.execution.is_none());
        assert_eq!(failed.source, file.source);
        Ok(())
    }

    #[test]
    fn validation_never_mutates_sources_actors_or_compile_cache() -> RuntimeResult<()> {
        let mut image = ImageApi::new();
        let file = image.write_source(
            "actors/checked",
            None,
            "pub fn handle(ctx, state, sender, message) { state }",
        )?;
        let actor = image.spawn(&file.path, Value::Integer(3), capacity(2))?;
        let before = image.inspect(actor)?;
        let artifacts = image.code.artifact_count();
        let sources = image.code.source_count();
        assert!(
            image
                .check_source("pub fn handle( {")?
                .iter()
                .any(|diagnostic| diagnostic.severity == runa_core::DiagnosticSeverity::Error)
        );
        assert!(image.check_source("pub fn main() { 1 }")?.is_empty());
        assert_eq!(image.source(&file.path), Some(&file));
        let after = image.inspect(actor)?;
        assert_eq!(after.implementation, before.implementation);
        assert_eq!(after.state, before.state);
        assert_eq!(after.status, before.status);
        assert_eq!(after.queued, before.queued);
        assert_eq!(after.reserved, before.reserved);
        assert_eq!(image.code.artifact_count(), artifacts);
        assert_eq!(image.code.source_count(), sources);
        Ok(())
    }

    #[test]
    fn source_rename_moves_history_and_guards_installed_and_pending_users() -> RuntimeResult<()> {
        let mut image = ImageApi::new();
        let first = image.write_source(
            "actors/first",
            None,
            "pub fn handle(ctx, state, sender, message) { state + 1 }",
        )?;
        let second = image.write_source(
            "actors/second",
            None,
            "pub fn handle(ctx, state, sender, message) { state + 2 }",
        )?;
        let commit = image.commit_source("actors/second")?;
        let actor = image.spawn("actors/first", Value::Integer(0), capacity(2))?;
        image.r#become(actor, "actors/second")?;
        assert!(
            image
                .delete_source("actors/first", first.source_id)
                .unwrap_err()
                .to_string()
                .contains("stop the actors first")
        );
        assert!(
            image
                .delete_source("actors/second", second.source_id)
                .is_err()
        );
        assert!(
            image
                .rename_source("actors/second", "nested/renamed", second.source_id, false)
                .is_err()
        );
        assert!(
            image
                .rename_source("actors/second", "actors/first", second.source_id, true)
                .is_err()
        );
        assert!(
            image
                .rename_source("actors/second", "nested/renamed", first.source_id, true)
                .is_err()
        );
        assert_eq!(image.pending_actor_path(actor), Some("actors/second"));
        image.rename_source("actors/second", "nested/renamed", second.source_id, true)?;
        assert_eq!(image.pending_actor_path(actor), Some("nested/renamed"));
        assert!(image.source("actors/second").is_none());
        assert_eq!(
            image
                .source_commit("nested/renamed", &commit.revision)
                .unwrap()
                .path,
            "nested/renamed"
        );
        image.tick(1, capacity(100));
        assert_eq!(image.actor_path(actor), Some("nested/renamed"));
        image.rename_source("nested/renamed", "renamed", second.source_id, true)?;
        assert_eq!(image.actor_path(actor), Some("renamed"));
        image.system(actor, SystemMessage::Kill)?;
        image.delete_source("renamed", second.source_id)?;
        assert!(image.source_history("renamed").is_none());
        Ok(())
    }

    #[test]
    fn temporary_behavior_is_actor_local_cas_guarded_and_reloadable() -> RuntimeResult<()> {
        let mut image = ImageApi::new();
        let first = image.write_source(
            "actors/counter",
            None,
            "pub fn handle(ctx, state, sender, message) { state + 1 }",
        )?;
        let actor = image.spawn("actors/counter", Value::Integer(0), capacity(2))?;
        let other = image.spawn("actors/counter", Value::Integer(0), capacity(2))?;
        let working = image.write_source(
            "actors/counter",
            Some(first.source_id),
            "pub fn handle(ctx, state, sender, message) { state + 10 }",
        )?;
        assert_eq!(image.actor_behavior(actor)?.source, first.source);
        assert!(
            image
                .write_actor_behavior(actor, first.source_id.to_string(), "pub fn handle( {")
                .is_err()
        );
        assert!(!image.has_temporary_behavior(actor));
        let temporary = image.write_actor_behavior(
            actor,
            first.source_id.to_string(),
            "pub fn handle(ctx, state, sender, message) { state + 100 }",
        )?;
        assert!(image.has_temporary_behavior(actor));
        assert!(!image.has_temporary_behavior(other));
        assert_eq!(
            image.source("actors/counter").unwrap().source,
            working.source
        );
        assert!(
            image
                .write_actor_behavior(actor, first.source_id.to_string(), &first.source)
                .is_err()
        );
        assert!(
            image
                .update_behavior_source(actor, first.source_id, &temporary.source_id.to_string())
                .is_err()
        );
        assert!(
            image
                .update_behavior_source(actor, working.source_id, &first.source_id.to_string())
                .is_err()
        );
        for id in [actor, other] {
            image.send(id, Value::Null)?;
        }
        image.tick(10, capacity(100));
        assert_eq!(image.inspect(actor)?.state, Value::Integer(100));
        assert_eq!(image.inspect(other)?.state, Value::Integer(1));
        let updated = image.update_behavior_source(
            actor,
            working.source_id,
            &temporary.source_id.to_string(),
        )?;
        assert_eq!(updated.source, temporary.source);
        assert!(image.source_history("actors/counter").unwrap().is_empty());
        image.write_source("actors/counter", Some(updated.source_id), &working.source)?;
        image.reload(actor)?;
        assert!(!image.has_temporary_behavior(actor));
        image.send(actor, Value::Null)?;
        image.tick(10, capacity(100));
        assert_eq!(image.inspect(actor)?.state, Value::Integer(110));
        assert_eq!(image.actor_behavior(actor)?.source, working.source);
        Ok(())
    }

    #[test]
    fn temporary_install_waits_for_active_handler_and_survives_restart() -> RuntimeResult<()> {
        let mut image = ImageApi::new();
        let file = image.write_source("actors/slow", None, "pub fn handle(ctx, state, sender, message) { let n = 0; while n < 100 { n += 1; } state + 1 }")?;
        let actor = image.spawn("actors/slow", Value::Integer(0), capacity(2))?;
        image.send(actor, Value::Null)?;
        image.tick(1, capacity(1));
        let active = image.inspect(actor)?;
        assert!(active.execution.is_some());
        image.write_actor_behavior(
            actor,
            file.source_id.to_string(),
            "pub fn handle(ctx, state, sender, message) { state + 10 }",
        )?;
        assert_eq!(image.inspect(actor)?.implementation, active.implementation);
        image.tick(10, capacity(1000));
        assert_eq!(image.inspect(actor)?.state, Value::Integer(1));
        image.send(actor, Value::Null)?;
        image.restart(actor)?;
        image.tick(10, capacity(1000));
        assert_eq!(image.inspect(actor)?.state, Value::Integer(11));
        image.system(actor, SystemMessage::Kill)?;
        assert!(!image.has_temporary_behavior(actor));
        Ok(())
    }

    #[test]
    fn failed_actors_can_repair_behavior_without_losing_reserved_messages() -> RuntimeResult<()> {
        let mut image = ImageApi::new();
        let file = image.write_source(
            "actors/broken",
            None,
            "pub fn handle(ctx, state, sender, message) { panic(\"broken\") }",
        )?;
        let actor = image.spawn("actors/broken", Value::Integer(0), capacity(2))?;
        image.send(actor, Value::Integer(7))?;
        image.tick(10, capacity(100));
        let before = image.inspect(actor)?;
        assert!(matches!(before.status, runa_actor::Status::Failed(_)));
        image.write_actor_behavior(
            actor,
            file.source_id.to_string(),
            "pub fn handle(ctx, state, sender, message) { state + message }",
        )?;
        assert!(matches!(
            image.inspect(actor)?.status,
            runa_actor::Status::Failed(_)
        ));
        assert_eq!(image.inspect(actor)?.reserved, before.reserved);
        image.restart(actor)?;
        image.tick(10, capacity(100));
        assert_eq!(image.inspect(actor)?.state, Value::Integer(7));
        Ok(())
    }

    #[test]
    fn invalid_uncommitted_source_fails_spawn_without_creating_actor() -> RuntimeResult<()> {
        let mut image = ImageApi::new();
        image.write_source("actors/broken", None, "pub fn handle( {")?;
        assert!(image.commit_source("actors/broken").is_ok());
        assert!(
            image
                .spawn("actors/broken", Value::Null, capacity(2))
                .is_err()
        );
        assert!(image.actors().is_empty());
        Ok(())
    }

    #[test]
    fn compatibility_behavior_seeds_only_its_exact_working_path() -> RuntimeResult<()> {
        let mut image = ImageApi::new();
        image.publish_behavior(
            "counter",
            None,
            "pub fn handle(ctx, state, sender, message) { state + message.value }",
        )?;
        assert!(image.source("counter").is_some());
        assert!(image.source("actors/counter").is_none());
        Ok(())
    }

    #[test]
    fn restarting_pending_become_preserves_the_installed_path() -> RuntimeResult<()> {
        let mut image = ImageApi::new();
        image.write_source(
            "actors/first",
            None,
            "pub fn handle(ctx, state, sender, message) { state + 1 }",
        )?;
        image.write_source(
            "actors/second",
            None,
            "pub fn handle(ctx, state, sender, message) { state + 10 }",
        )?;
        let actor = image.spawn("actors/first", Value::Integer(0), capacity(2))?;
        let original = image.inspect(actor)?.implementation;
        image.r#become(actor, "actors/second")?;
        assert_eq!(image.actor_paths[&actor], "actors/first");
        image.restart(actor)?;
        assert_eq!(image.reload(actor)?, original);
        image.tick(1, capacity(100));
        assert_eq!(image.inspect(actor)?.implementation, original);
        assert_eq!(image.actor_paths[&actor], "actors/first");

        image.r#become(actor, "actors/second")?;
        image.tick(1, capacity(100));
        assert_eq!(image.actor_paths[&actor], "actors/second");
        let current = image.inspect(actor)?.implementation;
        assert_eq!(image.reload(actor)?, current);
        image.system(actor, SystemMessage::Kill)?;
        assert!(image.inspect(actor).is_err());
        assert!(!image.actor_paths.contains_key(&actor));
        assert!(!image.pending_paths.contains_key(&actor));
        assert!(image.reload(actor).is_err());
        Ok(())
    }

    #[test]
    fn restart_forces_a_new_incarnation_during_active_execution() -> RuntimeResult<()> {
        let mut image = ImageApi::new();
        image.write_source(
            "actors/loop",
            None,
            "pub fn handle(ctx, state, sender, message) { loop {} }",
        )?;
        let actor = image.spawn("actors/loop", Value::Null, capacity(2))?;
        image.send(actor, Value::Null)?;
        image.tick(1, capacity(100));
        let before = image.inspect(actor)?;
        assert!(before.execution.is_some());
        image.restart(actor)?;
        let after = image.inspect(actor)?;
        assert_eq!(after.incarnation, before.incarnation + 1);
        assert_eq!(after.state, before.state);
        assert!(after.execution.is_none());
        Ok(())
    }

    #[test]
    fn reload_compiles_latest_working_source_and_preserves_state_and_mailbox() -> RuntimeResult<()>
    {
        let mut image = ImageApi::new();
        let file = image.write_source(
            "actors/counter",
            None,
            "pub fn handle(ctx, state, sender, message) { state + message }",
        )?;
        image.commit_source("actors/counter")?;
        let actor = image.spawn("actors/counter", Value::Integer(3), capacity(2))?;
        image.send(actor, Value::Integer(2))?;
        let before = image.inspect(actor)?;
        image.write_source(
            "actors/counter",
            Some(file.source_id),
            "pub fn handle(ctx, state, sender, message) { state + message * 10 }",
        )?;
        let implementation = image.reload(actor)?;
        let scheduled = image.inspect(actor)?;
        assert_eq!(scheduled.implementation, before.implementation);
        assert_eq!(scheduled.state, before.state);
        assert_eq!(scheduled.queued, before.queued);
        image.tick(100, capacity(100));
        let after = image.inspect(actor)?;
        assert_eq!(after.implementation, implementation);
        assert_eq!(after.state, Value::Integer(23));
        assert_eq!(after.incarnation, before.incarnation);
        assert_eq!(after.mailbox, before.mailbox);
        assert_eq!(image.actor_paths[&actor], "actors/counter");
        Ok(())
    }

    #[test]
    fn become_binds_only_after_install_and_reload_follows_the_new_path() -> RuntimeResult<()> {
        let mut image = ImageApi::new();
        image.write_source(
            "actors/first",
            None,
            "pub fn handle(ctx, state, sender, message) { state + message }",
        )?;
        let second = image.write_source(
            "actors/second",
            None,
            "pub fn handle(ctx, state, sender, message) { state * message }",
        )?;
        let actor = image.spawn("actors/first", Value::Integer(3), capacity(2))?;
        image.send(actor, Value::Integer(2))?;
        let before = image.inspect(actor)?;
        let implementation = image.r#become(actor, "actors/second")?;
        assert_eq!(image.actor_paths[&actor], "actors/first");
        assert_eq!(image.inspect(actor)?.queued, before.queued);
        image.tick(100, capacity(100));
        assert_eq!(image.actor_paths[&actor], "actors/second");
        assert_eq!(image.inspect(actor)?.implementation, implementation);
        assert_eq!(image.inspect(actor)?.state, Value::Integer(6));
        image.write_source(
            "actors/second",
            Some(second.source_id),
            "pub fn handle(ctx, state, sender, message) { state * message * 10 }",
        )?;
        let updated = image.reload(actor)?;
        assert_ne!(updated, implementation);
        image.send(actor, Value::Integer(2))?;
        image.tick(100, capacity(100));
        assert_eq!(image.inspect(actor)?.implementation, updated);
        assert_eq!(image.inspect(actor)?.state, Value::Integer(120));
        assert_eq!(image.inspect(actor)?.mailbox, before.mailbox);
        Ok(())
    }

    #[test]
    fn restart_and_superseding_reload_cancel_pending_become_path_changes() -> RuntimeResult<()> {
        let mut image = ImageApi::new();
        image.write_source(
            "actors/first",
            None,
            "pub fn handle(ctx, state, sender, message) { state + message }",
        )?;
        image.write_source(
            "actors/second",
            None,
            "pub fn handle(ctx, state, sender, message) { state * message }",
        )?;
        let actor = image.spawn("actors/first", Value::Integer(3), capacity(2))?;
        let before = image.inspect(actor)?;
        image.r#become(actor, "actors/second")?;
        image.restart(actor)?;
        image.tick(1, capacity(100));
        assert_eq!(image.actor_paths[&actor], "actors/first");
        assert_eq!(image.inspect(actor)?.implementation, before.implementation);
        assert_eq!(image.inspect(actor)?.incarnation, before.incarnation + 1);
        assert!(!image.pending_paths.contains_key(&actor));

        image.r#become(actor, "actors/second")?;
        assert_eq!(image.reload(actor)?, before.implementation);
        image.tick(1, capacity(100));
        assert_eq!(image.actor_paths[&actor], "actors/first");
        assert_eq!(image.inspect(actor)?.implementation, before.implementation);
        image.write_source("actors/broken", None, "pub fn broken( {")?;
        assert!(image.r#become(actor, "actors/broken").is_err());
        assert_eq!(image.actor_paths[&actor], "actors/first");
        assert!(image.r#become(actor, "actors/missing").is_err());
        assert_eq!(image.actor_paths[&actor], "actors/first");
        Ok(())
    }
}