spawn.rs

//! Actor spawn subsystem: creates new actors with state, bytecode handlers,
//! and recovery metadata. These free functions take `&mut Runtime` to access
//! the runtime's public fields.

use std::collections::HashMap;

use crate::runtime::actor::{Actor, ActorBackend, BehaviorEntry};
use crate::runtime::persistence::{PersistedValue, StateModel, WorkflowEvent};
use crate::runtime::timer_fired_handler;
use crate::runtime::Runtime;
use crate::runtime::{bytecode_step_placeholder, fresh_actor_id, map_ast_state_model};
use crate::vm::Value;

/// Core spawn logic shared by all spawn entry points.
pub(crate) fn spawn_actor_with_models(
    rt: &mut Runtime,
    init: Box<dyn FnOnce() -> Vec<(String, Value)>>,
    state_models: HashMap<String, StateModel>,
    persistent: bool,
    workflow: Option<&str>,
) -> u64 {
    let id = fresh_actor_id();
    let mut actor = Actor::new(id, format!("actor_{}", id), 0);
    let state_fields = init();
    for (name, value) in state_fields {
        actor.set_state_field(name, value);
    }
    actor.state_models = state_models;
    // Register CRDT-backed fields with the CrdtManager.
    for (field_name, model) in &actor.state_models {
        if let StateModel::Crdt(crdt_type) = model {
            let initial = actor
                .state_data
                .get(field_name)
                .copied()
                .unwrap_or(Value::nil());
            if let Some(ref mut mgr) = rt.crdt_manager {
                mgr.register_actor_field(id, field_name, *crdt_type, initial);
            }
        }
    }
    actor.persistent = persistent;
    let workflow_name = workflow.map(|n| n.to_string());
    if let Some(name) = workflow {
        actor.is_workflow = true;
        actor.name = name.to_string();
        actor.register_behavior("__timer_fired", timer_fired_handler);
    }
    actor.state = crate::runtime::ActorState::Running;
    rt.actors.insert(id, actor);
    if workflow.is_some() {
        let seq = crate::runtime::workflow::next_sequence(rt, id);
        let state = {
            let actor = rt.actors.get(&id).unwrap();
            let mut state = Vec::new();
            for (field_name, value) in &actor.state_data {
                let model = actor
                    .state_models
                    .get(field_name)
                    .copied()
                    .unwrap_or(StateModel::Local);
                if model.is_persistent() {
                    state.push(PersistedValue::from_value_resolved(
                        value,
                        actor.bytecode_module.as_ref(),
                    ));
                }
            }
            state
        };
        let _ = rt.persistence.append_workflow_event(
            id,
            WorkflowEvent::WorkflowStarted {
                sequence: seq,
                name: workflow_name.as_ref().unwrap().clone(),
                state,
            },
        );
        crate::runtime::workflow::checkpoint_actor(rt, id);
    }
    rt.enqueue_actor(id);
    id
}

/// Spawn an actor for `module`'s behavior `behavior_idx`, seeded with the
/// `init` state fields, and wire up its bytecode handlers. Shared body of
/// both VM-callback `spawn_actor` impls.
pub(crate) fn spawn_from_module(
    rt: &mut Runtime,
    module: &crate::bytecode::CodeModule,
    behavior_idx: usize,
    init: Vec<(String, Value)>,
) -> Value {
    let meta = module
        .actor_metadata
        .iter()
        .find(|m| m.behavior_indices.contains(&behavior_idx));
    let id = if let Some(meta) = meta {
        let state_models: HashMap<String, StateModel> = meta
            .state_models
            .iter()
            .map(|(name, model)| (name.clone(), map_ast_state_model(*model)))
            .collect();
        let defaults = meta.state_defaults.clone();
        spawn_actor_with_models(
            rt,
            Box::new(move || {
                let mut fields: Vec<(String, Value)> = defaults
                    .iter()
                    .map(|(name, c)| (name.clone(), crate::vm::constant_to_value(c)))
                    .collect();
                fields.extend(init);
                fields
            }),
            state_models,
            meta.persistent,
            if meta.is_workflow {
                Some(meta.name.as_str())
            } else {
                None
            },
        )
    } else {
        spawn_actor_with_models(rt, Box::new(move || init), HashMap::new(), false, None)
    };
    let offsets: Vec<usize> = module.behaviors.iter().map(|b| b.code_offset).collect();
    // compensation_offsets filtered to this actor's own behaviors so
    // step-local indices in run_saga_compensation match.
    let compensation_offsets: Vec<Option<usize>> = if let Some(meta) = meta {
        meta.behavior_indices
            .iter()
            .map(|&i| module.behaviors[i].compensate_offset)
            .collect()
    } else {
        module
            .behaviors
            .iter()
            .map(|b| b.compensate_offset)
            .collect()
    };
    if let Some(actor) = rt.actors.get_mut(&id) {
        actor.bytecode_module = Some(module.clone());
        actor.bytecode_offsets = offsets.clone();
        actor.compensation_offsets = compensation_offsets.clone();
        if let Some(meta) = meta {
            if meta.is_agent {
                actor.is_agent = true;
                for (name, c) in &meta.state_defaults {
                    if let crate::bytecode::Constant::String(json) = c {
                        if name == "retry_config" {
                            actor.retry_config = serde_json::from_str(json).ok();
                        } else if name == "fallback_config" {
                            actor.fallback_config = serde_json::from_str(json).unwrap_or_default();
                        }
                    }
                }
            }
            for (name, c) in &meta.state_defaults {
                if let crate::bytecode::Constant::String(s) = c {
                    let ptr = actor.allocate_string(s);
                    actor.set_state_field(name, ptr);
                }
            }
            actor.backend = match meta.backend {
                crate::ast::ActorBackendKind::Native => ActorBackend::Native,
                crate::ast::ActorBackendKind::WasmComponent => ActorBackend::WasmComponent {
                    component_path: String::new(),
                },
            };
        }
    }
    // Wire AOT-native dispatch: if an AOT module is registered for this actor
    // type, register the adapter for each behavior it compiles so the
    // scheduler dispatches them natively (bytecode falls back for the rest).
    if let Some(meta) = meta.as_ref() {
        if !meta.is_workflow {
            let module_ptr = rt.aot_modules.get(&meta.name).copied();
            if let Some(module_ptr) = module_ptr {
                let aot_module = unsafe { &*module_ptr };
                let runtime_ptr = rt as *mut Runtime;
                if let Some(actor) = rt.actors.get_mut(&id) {
                    for &gidx in &meta.behavior_indices {
                        // CodeModule behavior names are fully-qualified
                        // `"{Actor}.{behavior}"` (see mir_lower), which is
                        // exactly what `fn_ptr_for_behavior` expects.
                        let fq = module
                            .behaviors
                            .get(gidx)
                            .map(|b| b.name.clone())
                            .unwrap_or_default();
                        let short = fq
                            .strip_prefix(&format!("{}.", meta.name))
                            .map(str::to_string)
                            .unwrap_or_else(|| fq.clone());
                        if let Some(fn_ptr) = aot_module.fn_ptr_for_behavior(&fq) {
                            actor.register_behavior(short, crate::aot::aot_behavior_adapter);
                            actor.aot_targets.push(Some(crate::aot::AotDispatchTarget {
                                fn_ptr,
                                module: module_ptr,
                                runtime: runtime_ptr,
                            }));
                        } else {
                            actor.register_behavior(String::new(), bytecode_step_placeholder);
                            actor.aot_targets.push(None);
                        }
                    }
                }
            }
        }
    }
    if meta.map(|m| m.is_workflow).unwrap_or(false) {
        layout_workflow_behavior_table(rt, id);
    }
    register_recovery_module(rt, id, module.clone(), offsets, compensation_offsets);
    Value::actor_ref(id)
}

/// Populate a workflow actor's behavior table with placeholder entries for
/// each bytecode step plus the internal `__timer_fired` handler.
pub(crate) fn layout_workflow_behavior_table(rt: &mut Runtime, actor_id: u64) {
    if let Some(actor) = rt.actors.get_mut(&actor_id) {
        if !actor.is_workflow {
            return;
        }
        let step_count = actor.bytecode_offsets.len();
        actor
            .behavior_table
            .retain(|e| !e.name.is_empty() && e.name != "__timer_fired");
        for _ in 0..step_count {
            actor.behavior_table.push(BehaviorEntry {
                name: String::new(),
                handler_fn: bytecode_step_placeholder,
            });
        }
        actor.register_behavior("__timer_fired", timer_fired_handler);
    }
}

/// Register bytecode metadata so that a persistent actor can be recovered
/// after a runtime restart.
pub(crate) fn register_recovery_module(
    rt: &mut Runtime,
    actor_id: u64,
    module: crate::bytecode::CodeModule,
    offsets: Vec<usize>,
    compensation_offsets: Vec<Option<usize>>,
) {
    rt.recovery_modules
        .insert(actor_id, (module, offsets, compensation_offsets));
}