epics-base-rs 0.18.3

//! IOC Application — st.cmd-style startup for Rust IOCs.
//!
//! Provides a 2-phase IOC lifecycle matching the C++ EPICS pattern:
//!
//! **Phase 1 (pre-init):** Execute startup script (`st.cmd`)
//!   - `epicsEnvSet`, `dbLoadRecords`, custom driver config commands
//!
//! **Phase 2 (iocInit):** Wire device support, start protocol server
//!
//! **Phase 3 (post-init):** Interactive iocsh REPL
//!   - `dbl`, `dbgf`, `dbpf`, `dbpr`, custom commands
//!
//! # Example
//!
//! ```rust,ignore
//! IocApplication::new()
//!     .port(5064)
//!     .register_device_support("myDevice", || Box::new(MyDeviceSupport::new()))
//!     .register_startup_command(my_config_command())
//!     .startup_script("st.cmd")
//!     .run(my_protocol_runner)
//!     .await
//! ```

use std::collections::HashMap;
use std::sync::{Arc, Mutex};

use crate::error::{CaError, CaResult};
use crate::runtime::net::cas_server_port;
use crate::server::record::{self, Record, SubroutineFn};

use crate::server::database::PvDatabase;
use crate::server::device_support::DeviceSupport;
use crate::server::iocsh::{self, registry::CommandDef};
use crate::server::{DeviceSupportFactory, access_security, autosave};
use autosave::startup::AutosaveStartupConfig;

/// IOC lifecycle init-hook subsystem — Rust port of epics-base
/// `libcom/src/iocsh/initHooks.{c,h}`.
///
/// C code registers a callback via `initHookRegister()` and the IOC
/// fires `initHookAnnounce(state)` at fixed points during
/// `iocBuild()` / `iocRun()`. Ported code (autosave pass-0/pass-1
/// restore, areaDetector plugins, sequencer programs, caPutLog,
/// devIocStats) all hang behaviour off these announcements.
///
/// Both Rust build paths ([`IocApplication::run`] and
/// [`crate::server::ioc_builder::IocBuilder::build`]) announce the
/// states they reach in the same order C does.
pub mod init_hooks {
    use std::sync::{Arc, Mutex};

    /// Initialization stages, mirroring C's `initHookState` enum
    /// (`initHooks.h`). Only the states an embedded-style Rust IOC
    /// can actually reach are modelled; the C `iocPause` /
    /// `iocShutdown` / unit-test states are omitted because neither
    /// Rust build path has a pause/shutdown transition that announces
    /// them. The order of the modelled variants matches C exactly.
    #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
    pub enum InitHookState {
        /// Start of iocBuild() / iocInit().
        AtIocBuild,
        /// Database sanity checks passed.
        AtBeginning,
        /// Callbacks, generalTime & taskwd init.
        AfterCallbackInit,
        /// CA links init.
        AfterCaLinkInit,
        /// Driver support init.
        AfterInitDrvSup,
        /// Record support init.
        AfterInitRecSup,
        /// Device support init pass 0 (also autosave pass 0).
        AfterInitDevSup,
        /// Records and locksets init (also autosave pass 1).
        AfterInitDatabase,
        /// Device support init pass 1.
        AfterFinishDevSup,
        /// Scan, AS, ProcessNotify init.
        AfterScanInit,
        /// Records with PINI = YES processed.
        AfterInitialProcess,
        /// RSRV (CA server) init.
        AfterCaServerInit,
        /// End of iocBuild().
        AfterIocBuilt,
        /// Start of iocRun().
        AtIocRun,
        /// Scan tasks and CA links running.
        AfterDatabaseRunning,
        /// RSRV (CA server) running.
        AfterCaServerRunning,
        /// End of iocRun() / iocInit().
        AfterIocRunning,
    }

    impl InitHookState {
        /// Printable representation — mirrors C `initHookName()`.
        pub fn name(&self) -> &'static str {
            match self {
                InitHookState::AtIocBuild => "initHookAtIocBuild",
                InitHookState::AtBeginning => "initHookAtBeginning",
                InitHookState::AfterCallbackInit => "initHookAfterCallbackInit",
                InitHookState::AfterCaLinkInit => "initHookAfterCaLinkInit",
                InitHookState::AfterInitDrvSup => "initHookAfterInitDrvSup",
                InitHookState::AfterInitRecSup => "initHookAfterInitRecSup",
                InitHookState::AfterInitDevSup => "initHookAfterInitDevSup",
                InitHookState::AfterInitDatabase => "initHookAfterInitDatabase",
                InitHookState::AfterFinishDevSup => "initHookAfterFinishDevSup",
                InitHookState::AfterScanInit => "initHookAfterScanInit",
                InitHookState::AfterInitialProcess => "initHookAfterInitialProcess",
                InitHookState::AfterCaServerInit => "initHookAfterCaServerInit",
                InitHookState::AfterIocBuilt => "initHookAfterIocBuilt",
                InitHookState::AtIocRun => "initHookAtIocRun",
                InitHookState::AfterDatabaseRunning => "initHookAfterDatabaseRunning",
                InitHookState::AfterCaServerRunning => "initHookAfterCaServerRunning",
                InitHookState::AfterIocRunning => "initHookAfterIocRunning",
            }
        }
    }

    /// Application callback type — Rust equivalent of C's
    /// `initHookFunction`. Invoked once per announced state. `Arc`
    /// so [`init_hook_announce`] can snapshot the list and drop the
    /// lock before invoking callbacks (C holds its list mutex only
    /// during traversal, never across the callback).
    pub type InitHookFunction = Arc<dyn Fn(InitHookState) + Send + Sync>;

    static HOOKS: Mutex<Vec<InitHookFunction>> = Mutex::new(Vec::new());

    /// Register a function for initHook notifications — Rust port of
    /// C `initHookRegister()`. The callback is invoked for every
    /// subsequently-announced state. Registration is process-global,
    /// matching C's single `functionList`.
    ///
    /// Unlike C (which dedups by function pointer) closures cannot be
    /// compared for identity, so every call adds a distinct callback;
    /// callers must register each hook once.
    pub fn init_hook_register(func: InitHookFunction) {
        HOOKS.lock().unwrap().push(func);
    }

    /// Announce an init-hook state to all registered callbacks —
    /// Rust port of C `initHookAnnounce()`. Called only by the IOC
    /// build paths at the fixed lifecycle points.
    ///
    /// The callback list is snapshotted (cheap `Arc` clones) and the
    /// lock released before any callback runs, so a hook that calls
    /// [`init_hook_register`] from inside the callback cannot
    /// deadlock. Hooks registered during an announce are not invoked
    /// for that same state — matching C's snapshot-of-`ellFirst`
    /// traversal semantics.
    pub fn init_hook_announce(state: InitHookState) {
        let snapshot: Vec<InitHookFunction> = HOOKS.lock().unwrap().clone();
        for cb in snapshot {
            cb(state);
        }
    }

    /// Forget all registered callbacks. Test-only — mirrors C
    /// `initHookFree()`. Lets unit tests run in isolation without
    /// leaking process-global hook state into each other.
    #[cfg(test)]
    pub fn init_hook_free() {
        HOOKS.lock().unwrap().clear();
    }
}

pub use init_hooks::{InitHookFunction, InitHookState, init_hook_announce, init_hook_register};

/// Context passed to dynamic device support factories during iocInit wiring.
pub struct DeviceSupportContext<'a> {
    pub dtyp: &'a str,
    pub inp: &'a str,
    pub out: &'a str,
}

/// Dynamic device support factory: given a context, returns device support if recognized.
pub type DynamicDeviceSupportFactory =
    Box<dyn Fn(&DeviceSupportContext) -> Option<Box<dyn DeviceSupport>> + Send + Sync>;

/// Configuration passed to the protocol runner after IOC initialization.
///
/// Contains all the pieces needed to start a protocol-specific server
/// (e.g., CA or PVA) with an interactive shell.
pub struct IocRunConfig {
    pub db: Arc<PvDatabase>,
    /// UDP discovery port — clients SEARCH here. Defaults to
    /// `EPICS_CA_SERVER_PORT` or 5064.
    pub port: u16,
    /// Optional TCP-listen port override. `None` means "use `port`".
    /// `Some(p)` lets multiple IOCs on one host bind unique TCP ports
    /// (epics-base PR #69, `EPICS_CAS_SERVER_PORT`) while keeping the
    /// canonical UDP discovery port.
    pub tcp_port: Option<u16>,
    pub acf: Option<access_security::AccessSecurityConfig>,
    pub autosave_config: Option<autosave::SaveSetConfig>,
    pub autosave_manager: Option<Arc<autosave::AutosaveManager>>,
    pub shell_commands: Vec<CommandDef>,
    /// Retained for API compatibility. [`IocApplication::run`] now
    /// drains `register_after_init` hooks itself at the
    /// `initHookAfterIocRunning` point, so this is always handed to
    /// the protocol runner EMPTY. A runner must not execute it
    /// (doing so is a no-op on the empty vec, but the hooks have
    /// already run).
    pub after_init_hooks: Vec<Box<dyn FnOnce() + Send>>,
}

/// IOC Application with st.cmd-style startup support.
pub struct IocApplication {
    port: u16,
    /// Optional TCP listen port override. `None` means "share with UDP
    /// discovery port". Set via [`Self::tcp_port`] or the
    /// `EPICS_CAS_SERVER_PORT` env var (resolved at run time).
    tcp_port: Option<u16>,
    device_factories: HashMap<String, DeviceSupportFactory>,
    dynamic_device_factory: Option<DynamicDeviceSupportFactory>,
    record_factories: HashMap<String, super::RecordFactory>,
    subroutine_registry: HashMap<String, Arc<SubroutineFn>>,
    acf: Option<access_security::AccessSecurityConfig>,
    autosave_config: Option<autosave::SaveSetConfig>,
    autosave_startup: Option<Arc<Mutex<AutosaveStartupConfig>>>,
    startup_commands: Vec<CommandDef>,
    shell_commands: Vec<CommandDef>,
    startup_script: Option<String>,
    /// Records added via the declarative builder (Phase 7).
    inline_records: Vec<(String, Box<dyn Record>)>,
    /// Callbacks invoked after iocInit completes (e.g., start pollers).
    after_init_hooks: Vec<Box<dyn FnOnce() + Send>>,
}

impl IocApplication {
    pub fn new() -> Self {
        let mut device_factories: HashMap<String, DeviceSupportFactory> = HashMap::new();
        // epics-base 3.15.4: built-in `getenv` device support for
        // stringin / lsi — pre-registered so .db files can use the
        // canonical DTYP name with zero extra setup.
        device_factories.insert(
            "getenv".to_string(),
            Box::new(|| -> Box<dyn DeviceSupport> {
                Box::new(crate::server::builtin_devices::GetenvDeviceSupport::new())
            }),
        );
        Self {
            // SERVER-side port: caservertask.c:491-498 honours
            // EPICS_CAS_SERVER_PORT > EPICS_CA_SERVER_PORT > 5064.
            port: cas_server_port(),
            tcp_port: None,
            device_factories,
            dynamic_device_factory: None,
            record_factories: HashMap::new(),
            subroutine_registry: HashMap::new(),
            acf: None,
            autosave_config: None,
            autosave_startup: None,
            startup_commands: Vec::new(),
            shell_commands: Vec::new(),
            startup_script: None,
            inline_records: Vec::new(),
            after_init_hooks: Vec::new(),
        }
    }

    /// Set the UDP discovery port (default: 5064).
    pub fn port(mut self, port: u16) -> Self {
        self.port = port;
        self
    }

    /// Set the TCP listen port independently from the UDP discovery
    /// port (epics-base PR #69, `EPICS_CAS_SERVER_PORT`). Multiple IOCs
    /// on one host can each bind a unique TCP port while sharing the
    /// canonical 5064 UDP search port. When unset, the IOC resolves it
    /// at run time from `EPICS_CAS_SERVER_PORT`; if that's also unset,
    /// the TCP listener inherits [`Self::port`].
    pub fn tcp_port(mut self, port: u16) -> Self {
        self.tcp_port = Some(port);
        self
    }

    /// Register a device support factory by DTYP name.
    /// Called during iocInit to wire device support to records.
    pub fn register_device_support<F>(mut self, dtyp: &str, factory: F) -> Self
    where
        F: Fn() -> Box<dyn DeviceSupport> + Send + Sync + 'static,
    {
        self.device_factories
            .insert(dtyp.to_string(), Box::new(factory));
        self
    }

    /// Register a dynamic device support factory.
    ///
    /// Called as a fallback when a record's DTYP doesn't match any
    /// statically registered factory. The closure receives the DTYP name
    /// and returns `Some(device_support)` if it can handle that DTYP.
    ///
    /// Multiple calls are chained: new factory is tried first, then existing.
    pub fn register_dynamic_device_support<F>(mut self, factory: F) -> Self
    where
        F: Fn(&DeviceSupportContext) -> Option<Box<dyn DeviceSupport>> + Send + Sync + 'static,
    {
        if let Some(existing) = self.dynamic_device_factory.take() {
            self.dynamic_device_factory = Some(Box::new(move |ctx: &DeviceSupportContext| {
                factory(ctx).or_else(|| existing(ctx))
            }));
        } else {
            self.dynamic_device_factory = Some(Box::new(factory));
        }
        self
    }

    /// Register a command available during startup script execution (Phase 1).
    /// Use this for driver configuration commands like `simDetectorConfig`.
    pub fn register_startup_command(mut self, cmd: CommandDef) -> Self {
        self.startup_commands.push(cmd);
        self
    }

    /// Register a command available in the interactive shell (Phase 3).
    /// Use this for runtime commands like `simDetectorReport`.
    pub fn register_shell_command(mut self, cmd: CommandDef) -> Self {
        self.shell_commands.push(cmd);
        self
    }

    /// Register a callback to run after iocInit completes.
    ///
    /// Use this to start pollers and other periodic tasks that should
    /// not run during st.cmd execution or autosave restore.
    ///
    /// [`Self::run`] guarantees these fire — they are drained inside
    /// `run` at the `initHookAfterIocRunning` point (after PINI
    /// processing, before handoff to the protocol runner). They are
    /// NOT delegated to the protocol runner, so a custom runner does
    /// not need to remember to drain them.
    pub fn register_after_init(mut self, hook: impl FnOnce() + Send + 'static) -> Self {
        self.after_init_hooks.push(Box::new(hook));
        self
    }

    /// Set the startup script path (executed before iocInit).
    pub fn startup_script(mut self, path: &str) -> Self {
        self.startup_script = Some(path.to_string());
        self
    }

    /// Register a record type factory (e.g., "motor", "asyn").
    /// Avoids the global registry — factories are passed to IocBuilder.
    pub fn register_record_type<F>(mut self, type_name: &str, factory: F) -> Self
    where
        F: Fn() -> Box<dyn Record> + Send + Sync + 'static,
    {
        self.record_factories
            .insert(type_name.to_string(), Box::new(factory));
        self
    }

    /// Register a subroutine function by name (for sub records).
    pub fn register_subroutine<F>(mut self, name: &str, func: F) -> Self
    where
        F: Fn(&mut dyn Record) -> CaResult<()> + Send + Sync + 'static,
    {
        self.subroutine_registry
            .insert(name.to_string(), Arc::new(Box::new(func)));
        self
    }

    /// Configure autosave with a save set configuration.
    pub fn autosave(mut self, config: autosave::SaveSetConfig) -> Self {
        self.autosave_config = Some(config);
        self
    }

    /// Configure autosave startup (C-compatible iocsh commands).
    ///
    /// When set, autosave iocsh commands (`set_requestfile_path`, `create_monitor_set`,
    /// `set_pass0_restoreFile`, etc.) are registered as startup commands and populate
    /// the config during st.cmd execution. After iocInit, the config is consumed to
    /// build an `AutosaveManager`.
    pub fn autosave_startup(mut self, config: Arc<Mutex<AutosaveStartupConfig>>) -> Self {
        self.autosave_startup = Some(config);
        self
    }

    /// Configure access security.
    pub fn acf(mut self, config: access_security::AccessSecurityConfig) -> Self {
        self.acf = Some(config);
        self
    }

    // --- Declarative IOC Builder (Phase 7) ---

    /// Add a typed record to the IOC (no .db file needed).
    ///
    /// ```rust,ignore
    /// IocApplication::new()
    ///     .record("sensor:temp", AiRecord::new(0.0))
    ///     .record("heater:sp", AoRecord::new(0.0))
    ///     .run(my_runner).await
    /// ```
    pub fn record(mut self, name: &str, record: impl Record) -> Self {
        self.inline_records
            .push((name.to_string(), Box::new(record)));
        self
    }

    /// Add a pre-boxed record.
    pub fn record_boxed(mut self, name: &str, record: Box<dyn Record>) -> Self {
        self.inline_records.push((name.to_string(), record));
        self
    }

    /// Run the full IOC lifecycle: startup script -> iocInit -> protocol runner.
    ///
    /// The `protocol_runner` closure receives an [`IocRunConfig`] containing the
    /// fully initialized database, port, and configuration. It is responsible for
    /// starting the protocol-specific server (e.g., CA, PVA) and the interactive shell.
    pub async fn run<F, Fut>(self, protocol_runner: F) -> CaResult<()>
    where
        F: FnOnce(IocRunConfig) -> Fut + Send + 'static,
        Fut: std::future::Future<Output = CaResult<()>> + Send,
    {
        let db = Arc::new(PvDatabase::new());
        let handle = tokio::runtime::Handle::current();

        let Self {
            port,
            tcp_port,
            device_factories,
            dynamic_device_factory,
            record_factories,
            subroutine_registry,
            acf,
            autosave_config,
            autosave_startup,
            mut startup_commands,
            shell_commands,
            startup_script,
            inline_records,
            after_init_hooks,
        } = self;

        // Register record type factories with global registry so dbLoadRecords
        // (called from st.cmd) can find them. This bridges the injected factories
        // to the global registry that the iocsh dbLoadRecords command uses.
        for (name, factory) in record_factories {
            super::db_loader::register_record_type(&name, factory);
        }

        // Register autosave startup commands if configured
        if let Some(ref config) = autosave_startup {
            let cmds = AutosaveStartupConfig::register_startup_commands(config.clone());
            startup_commands.extend(cmds);
        }

        // Add inline records (Phase 7 declarative builder)
        for (name, record) in inline_records {
            db.add_record(&name, record).await?;
        }

        // Phase 1: Execute startup script in a separate std::thread.
        // std::thread (not spawn_blocking) is required because iocsh commands
        // use Handle::block_on() which panics inside the tokio runtime context.
        if let Some(script) = startup_script {
            let db1 = db.clone();
            let h1 = handle.clone();

            let (tx, rx) = crate::runtime::sync::oneshot::channel();
            std::thread::Builder::new()
                .name("iocsh-startup".into())
                .spawn(move || {
                    let shell = iocsh::IocShell::new(db1, h1);
                    for cmd in startup_commands {
                        shell.register(cmd);
                    }
                    let result = shell.execute_script(&script);
                    let _ = tx.send(result);
                })
                .expect("failed to spawn startup thread");

            let result = rx
                .await
                .map_err(|_| CaError::InvalidValue("startup thread dropped".into()))?;
            result.map_err(|e| CaError::InvalidValue(e))?;
        }

        // Collect restore paths and builder from startup config (scoped mutex lock)
        let (pass0_files, pass1_files, builder_opt) = if let Some(ref config) = autosave_startup {
            let cfg = config.lock().unwrap();
            let pass0: Vec<std::path::PathBuf> = cfg
                .pass0_restores
                .iter()
                .map(|r| cfg.resolve_save_file(&r.filename))
                .collect();
            let pass1: Vec<std::path::PathBuf> = cfg
                .pass1_restores
                .iter()
                .map(|r| cfg.resolve_save_file(&r.filename))
                .collect();
            let builder = if !cfg.monitor_sets.is_empty() || !cfg.triggered_sets.is_empty() {
                Some(cfg.into_builder())
            } else {
                None
            };
            (pass0, pass1, builder)
        } else {
            (Vec::new(), Vec::new(), None)
        };

        // initHooks subsystem (C `iocInit.c` / `initHooks.c` parity).
        //
        // Autosave pass-0 / pass-1 restore are no longer hard-coded
        // into the build flow: they are registered here as ordinary
        // init hooks (C autosave registers `initHookAfterInitDevSup`
        // for pass 0 and `initHookAfterInitDatabase` for pass 1).
        // Any third-party `init_hook_register` callback also fires at
        // the matching `init_hook_announce` point below. Because the
        // restore work is async and the C-parity `InitHookFunction`
        // is sync, autosave restores live in this local async-hook
        // table; `announce` below fires *both* tables.
        type AsyncHook = Box<
            dyn FnOnce()
                    -> std::pin::Pin<Box<dyn std::future::Future<Output = ()> + Send + 'static>>
                + Send
                + 'static,
        >;
        let mut lifecycle_hooks: Vec<(InitHookState, AsyncHook)> = Vec::new();

        // Register pass-0 restore as an `AfterInitDevSup` hook.
        {
            let db_p0 = db.clone();
            let files = pass0_files.clone();
            lifecycle_hooks.push((
                InitHookState::AfterInitDevSup,
                Box::new(move || {
                    Box::pin(async move {
                        for sav_path in &files {
                            match autosave::restore_from_file(&db_p0, sav_path).await {
                                Ok(count) if count > 0 => {
                                    eprintln!(
                                        "pass0 restore: {count} PVs from {}",
                                        sav_path.display()
                                    );
                                }
                                Err(e) => {
                                    eprintln!(
                                        "pass0 restore warning: {} - {e}",
                                        sav_path.display()
                                    );
                                }
                                _ => {}
                            }
                        }
                    })
                }),
            ));
        }
        // Register pass-1 restore + SaveSetConfig restore as an
        // `AfterInitDatabase` hook.
        {
            let db_p1 = db.clone();
            let files = pass1_files.clone();
            let cfg_path = autosave_config.as_ref().map(|c| c.save_path.clone());
            lifecycle_hooks.push((
                InitHookState::AfterInitDatabase,
                Box::new(move || {
                    Box::pin(async move {
                        for sav_path in &files {
                            match autosave::restore_from_file(&db_p1, sav_path).await {
                                Ok(count) if count > 0 => {
                                    eprintln!(
                                        "pass1 restore: {count} PVs from {}",
                                        sav_path.display()
                                    );
                                }
                                Err(e) => {
                                    eprintln!(
                                        "pass1 restore warning: {} - {e}",
                                        sav_path.display()
                                    );
                                }
                                _ => {}
                            }
                        }
                        if let Some(path) = cfg_path {
                            match autosave::restore_from_file(&db_p1, &path).await {
                                Ok(count) if count > 0 => {
                                    eprintln!("autosave: restored {count} PVs");
                                }
                                Err(e) => {
                                    eprintln!("autosave restore warning: {} - {e}", path.display());
                                }
                                _ => {}
                            }
                        }
                    })
                }),
            ));
        }

        // Fire an init-hook state: the C-parity sync `init_hook_*`
        // callbacks first, then the local async lifecycle hooks
        // (autosave restore). Drains every lifecycle hook matching
        // `state` out of the table so each fires exactly once.
        macro_rules! announce {
            ($state:expr) => {{
                let state = $state;
                init_hook_announce(state);
                let mut i = 0;
                while i < lifecycle_hooks.len() {
                    if lifecycle_hooks[i].0 == state {
                        let (_, hook) = lifecycle_hooks.remove(i);
                        hook().await;
                    } else {
                        i += 1;
                    }
                }
            }};
        }

        // iocBuild begins.
        announce!(InitHookState::AtIocBuild);
        announce!(InitHookState::AtBeginning);
        announce!(InitHookState::AfterCallbackInit);
        announce!(InitHookState::AfterCaLinkInit);
        announce!(InitHookState::AfterInitDrvSup);
        announce!(InitHookState::AfterInitRecSup);

        // Phase 2b: iocInit — wire device support to all records with DTYP.
        // C: initDevSup() then initHookAfterInitDevSup (autosave pass 0).
        let record_count =
            wire_device_support(&db, &device_factories, &dynamic_device_factory).await?;
        announce!(InitHookState::AfterInitDevSup);
        wire_subroutines(&db, &subroutine_registry).await;
        let io_intr_count = setup_io_intr(db.clone()).await;
        db.setup_cp_links().await;

        // Phase 2b.5: wait for external CA/PVA links to connect before
        // PINI runs (epics-base PR #768/#856 — `dbCa: iocInit wait`).
        // Default 10s timeout, override via `EPICS_RS_INIT_LINK_TIMEOUT`
        // (seconds, fractional accepted). Pass-through when no LinkSet
        // is registered.
        let link_wait_secs = crate::runtime::env::get("EPICS_RS_INIT_LINK_TIMEOUT")
            .and_then(|s| s.parse::<f64>().ok())
            .unwrap_or(10.0)
            .max(0.0);
        if link_wait_secs > 0.0 {
            let (connected, total) = db
                .wait_for_external_links(std::time::Duration::from_secs_f64(link_wait_secs))
                .await;
            if total > 0 {
                if connected == total {
                    eprintln!("iocInit: {connected}/{total} external links connected");
                } else {
                    eprintln!(
                        "iocInit: {connected}/{total} external links connected after \
                         {link_wait_secs}s — proceeding with unconnected links"
                    );
                }
            }
        }

        // C: initDatabase() then initHookAfterInitDatabase (autosave
        // pass 1). The registered hook performs pass-1 + SaveSetConfig
        // restore.
        announce!(InitHookState::AfterInitDatabase);
        announce!(InitHookState::AfterFinishDevSup);
        announce!(InitHookState::AfterScanInit);

        // Phase 2b.6: process PINI=YES records BEFORE the protocol
        // runner can accept client connections (H2 — match C's
        // iocBuild ordering: initialProcess() runs inside iocBuild,
        // before iocRun starts the CA server). Without this, a CA
        // client connecting in the first moments after IOC start
        // could `caget` a PINI record's UDF/default value instead of
        // its processed value. C guarantees this cannot happen.
        {
            let pini_records = db.pini_records().await;
            for name in &pini_records {
                let mut visited = std::collections::HashSet::new();
                let _ = db.process_record_with_links(name, &mut visited, 0).await;
            }
            // Publish completion so any scan scheduler started by the
            // protocol runner sees PINI as already done and its
            // non-owner branch does not block. The scheduler's owner
            // branch still re-runs the PINI burst; that is benign
            // (PINI records simply recompute) and avoids touching
            // `scan.rs`, which is outside this change's file scope.
            db.mark_pini_done();
        }
        announce!(InitHookState::AfterInitialProcess);

        // Phase 2d: Build AutosaveManager from startup config
        let autosave_manager = if let Some(builder) = builder_opt {
            match builder.build().await {
                Ok(mgr) => {
                    eprintln!("autosave: {} save set(s) configured", mgr.set_names().len());
                    Some(Arc::new(mgr))
                }
                Err(e) => {
                    eprintln!("autosave: failed to build manager: {e}");
                    None
                }
            }
        } else {
            None
        };

        let total_records = db.all_record_names().await.len();
        eprintln!(
            "iocInit: {total_records} records, {record_count} with device support, {io_intr_count} I/O Intr"
        );

        // C: rsrv init / iocBuild end. The Rust CA/PVA listener is
        // owned by the protocol runner, but PINI is already complete
        // (Phase 2b.6) so announcing here keeps hook ordering correct
        // for consumers that key off these states.
        announce!(InitHookState::AfterCaServerInit);
        announce!(InitHookState::AfterIocBuilt);
        // iocRun begins. Scan tasks / CA links are started by the
        // protocol runner immediately after handoff.
        announce!(InitHookState::AtIocRun);
        announce!(InitHookState::AfterDatabaseRunning);
        announce!(InitHookState::AfterCaServerRunning);

        // H3: drain `after_init_hooks` HERE — a guaranteed drain
        // point inside `run`. These were previously moved into
        // `IocRunConfig.after_init_hooks` and silently dropped
        // unless the external protocol runner remembered to execute
        // the vector. `register_after_init` promises "run after
        // iocInit completes"; PINI is done and the database is
        // built, so this is the correct C `initHookAfterIocRunning`
        // equivalent point. The `IocRunConfig.after_init_hooks`
        // field is now always handed over EMPTY (kept for API
        // compatibility) so a runner cannot double-run them.
        for hook in after_init_hooks {
            hook();
        }
        announce!(InitHookState::AfterIocRunning);

        // Phase 2e: drain `afterIocRunning` queue (epics-base PR #558).
        // Each line is an iocsh command queued by the startup script;
        // execute through a fresh shell so post-init state (including
        // PINI side effects) is visible. Both built-in iocsh commands
        // AND every user-registered `shell_commands` entry are
        // re-registered on this shell (CommandDef now `Clone`-able)
        // so site-specific names like `motorReport` are addressable
        // from the post-init queue.
        let pending = db.take_after_ioc_running();
        if !pending.is_empty() {
            let db1 = db.clone();
            let h1 = handle.clone();
            let shell_cmds_clone = shell_commands.clone();
            let (tx, rx) = crate::runtime::sync::oneshot::channel();
            std::thread::Builder::new()
                .name("iocsh-after-ioc-running".into())
                .spawn(move || {
                    let shell = iocsh::IocShell::new(db1, h1);
                    for cmd in shell_cmds_clone {
                        shell.register(cmd);
                    }
                    let mut errs: Vec<String> = Vec::new();
                    for line in pending {
                        if let Err(e) = shell.execute_line(&line) {
                            errs.push(format!("{line}: {e}"));
                        }
                    }
                    let _ = tx.send(errs);
                })
                .expect("failed to spawn afterIocRunning thread");
            if let Ok(errs) = rx.await {
                for e in errs {
                    eprintln!("afterIocRunning: {e}");
                }
            }
        }

        // Phase 3: Hand off to protocol runner.
        // C parity (caservertask.c:491-499): the server-side env var
        // EPICS_CAS_SERVER_PORT sets `ca_server_port`, and
        // `ca_udp_port = ca_server_port` — so UDP and TCP bind the
        // same value unless the Rust-extension `.tcp_port(...)`
        // explicitly splits them. The `port` field already
        // incorporates the CAS / CA / default precedence via
        // `cas_server_port()` (see `IocApplication::new`).
        // `tcp_port` here remains `Some(...)` only when the caller
        // explicitly invoked `.tcp_port(...)`; otherwise the runner
        // will inherit `port`.
        let config = IocRunConfig {
            db,
            port,
            tcp_port,
            acf,
            autosave_config,
            autosave_manager,
            shell_commands,
            // Already drained above (H3). Handed over empty so a
            // protocol runner that still inspects the field cannot
            // double-run the hooks.
            after_init_hooks: Vec::new(),
        };

        // epics-base PR #671 parity: race the protocol runner against
        // SIGTERM/SIGINT so a `kill` (or Ctrl+C on the controlling
        // terminal) cleanly returns Ok(()) instead of leaving the
        // future suspended forever. The CA/PVA runners already wire
        // their own signal handlers when used standalone; this one
        // covers the `IocApplication::run` entry point where the
        // runner closure may not (e.g., a custom user runner that
        // only sleeps on `pending()`).
        let runner_fut = protocol_runner(config);
        tokio::pin!(runner_fut);
        let ctrl_c = async {
            let _ = tokio::signal::ctrl_c().await;
        };
        #[cfg(unix)]
        let sigterm = async {
            if let Ok(mut sig) =
                tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate())
            {
                let _ = sig.recv().await;
            } else {
                std::future::pending::<()>().await;
            }
        };
        #[cfg(not(unix))]
        let sigterm = std::future::pending::<()>();

        tokio::select! {
            biased;
            // Runner takes priority: if it completes naturally
            // before any signal arrives, propagate its result.
            res = &mut runner_fut => res,
            _ = ctrl_c => {
                tracing::info!(target: "epics_base_rs::ioc_app", "SIGINT received, shutting down IOC");
                Ok(())
            }
            _ = sigterm => {
                tracing::info!(target: "epics_base_rs::ioc_app", "SIGTERM received, shutting down IOC");
                Ok(())
            }
        }
    }
}

/// Wire device support to all records that have DTYP set.
pub(crate) async fn wire_device_support(
    db: &PvDatabase,
    factories: &HashMap<String, DeviceSupportFactory>,
    dynamic_factory: &Option<DynamicDeviceSupportFactory>,
) -> CaResult<usize> {
    let names = db.all_record_names().await;
    let mut count = 0;
    for name in names {
        if let Some(rec_arc) = db.get_record(&name).await {
            let mut instance = rec_arc.write().await;
            let dtyp = instance.common.dtyp.clone();
            if !crate::server::device_support::is_soft_dtyp(&dtyp) {
                let ctx = DeviceSupportContext {
                    dtyp: &dtyp,
                    inp: &instance.common.inp,
                    out: &instance.common.out,
                };
                let dev_opt = if let Some(factory) = factories.get(&dtyp) {
                    Some(factory())
                } else if let Some(dyn_factory) = dynamic_factory {
                    dyn_factory(&ctx)
                } else {
                    None
                };
                if let Some(dev) = dev_opt {
                    // Canonical device-support init order (M1/M2):
                    // set_record_info → apply_record_info → init,
                    // with init-failure logged and the record flagged
                    // INVALID. Single owner of the contract, shared
                    // with the IocBuilder build path.
                    crate::server::device_support::wire_device_to_record(&mut instance, dev);
                    count += 1;
                } else {
                    eprintln!(
                        "warning: no device support registered for DTYP '{dtyp}' (record: {name})"
                    );
                }
            }
        }
    }
    Ok(count)
}

/// Wire subroutine functions to sub records.
async fn wire_subroutines(db: &PvDatabase, registry: &HashMap<String, Arc<SubroutineFn>>) {
    if registry.is_empty() {
        return;
    }
    let names = db.all_record_names().await;
    for name in names {
        if let Some(rec_arc) = db.get_record(&name).await {
            let mut instance = rec_arc.write().await;
            if instance.record.record_type() == "sub" {
                if let Some(crate::types::EpicsValue::String(snam)) =
                    instance.record.get_field("SNAM")
                {
                    if let Some(sub_fn) = registry.get(&snam) {
                        instance.subroutine = Some(sub_fn.clone());
                    }
                }
            }
        }
    }
}

/// Set up I/O Intr scanning for records with SCAN="I/O Intr".
async fn setup_io_intr(db: Arc<PvDatabase>) -> usize {
    let all_names = db.all_record_names().await;
    let io_intr_recs: Vec<(
        String,
        Arc<crate::runtime::sync::RwLock<record::RecordInstance>>,
    )> = {
        let mut recs = Vec::new();
        for name in &all_names {
            if let Some(arc) = db.get_record(name).await {
                recs.push((name.clone(), arc));
            }
        }
        recs
    };

    let mut count = 0;
    for (name, rec_arc) in io_intr_recs {
        let mut inst = rec_arc.write().await;
        if inst.common.scan == record::ScanType::IoIntr {
            if let Some(mut dev) = inst.device.take() {
                if let Some(mut intr_rx) = dev.io_intr_receiver() {
                    let db_clone = db.clone();
                    let rec_name = name.clone();
                    let rec_arc_clone = rec_arc.clone();
                    crate::runtime::task::spawn(async move {
                        while intr_rx.recv().await.is_some() {
                            // Only process if record is still on I/O Intr scan
                            let is_io_intr = {
                                let inst = rec_arc_clone.read().await;
                                inst.common.scan == record::ScanType::IoIntr
                            };
                            if !is_io_intr {
                                continue;
                            }
                            let mut visited = std::collections::HashSet::new();
                            let _ = db_clone
                                .process_record_with_links(&rec_name, &mut visited, 0)
                                .await;
                        }
                    });
                    count += 1;
                }
                inst.device = Some(dev);
            }
        }
    }
    count
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::sync::Mutex as StdMutex;
    use std::sync::atomic::{AtomicUsize, Ordering};

    /// Serialises the initHooks tests — `HOOKS` is process-global, so
    /// two tests announcing at once would observe each other's
    /// callbacks. The state machine here is small; a mutex is enough.
    static INIT_HOOK_TEST_LOCK: StdMutex<()> = StdMutex::new(());

    /// H1 regression: a callback registered via `init_hook_register`
    /// fires for every announced state, and `init_hook_announce`
    /// delivers states in the order they were announced.
    #[test]
    fn init_hook_register_and_announce_in_order() {
        let _guard = INIT_HOOK_TEST_LOCK.lock().unwrap();
        init_hooks::init_hook_free();

        let seen: Arc<StdMutex<Vec<InitHookState>>> = Arc::new(StdMutex::new(Vec::new()));
        let seen_cb = seen.clone();
        init_hook_register(Arc::new(move |state| {
            seen_cb.lock().unwrap().push(state);
        }));

        // Announce a subset in C order.
        let order = [
            InitHookState::AtIocBuild,
            InitHookState::AfterInitDevSup,
            InitHookState::AfterInitDatabase,
            InitHookState::AfterInitialProcess,
            InitHookState::AfterIocRunning,
        ];
        for &s in &order {
            init_hook_announce(s);
        }

        let got = seen.lock().unwrap().clone();
        assert_eq!(got, order, "hooks must fire in announce order");

        init_hooks::init_hook_free();
    }

    /// H1 regression: a hook that registers ANOTHER hook from inside
    /// its callback must not deadlock, and the newly-registered hook
    /// is not invoked for the in-progress state (C snapshot
    /// semantics).
    #[test]
    fn init_hook_reentrant_register_does_not_deadlock() {
        let _guard = INIT_HOOK_TEST_LOCK.lock().unwrap();
        init_hooks::init_hook_free();

        let inner_calls = Arc::new(AtomicUsize::new(0));
        let inner_for_outer = inner_calls.clone();
        init_hook_register(Arc::new(move |_state| {
            // Register a second hook from inside the callback.
            let inner = inner_for_outer.clone();
            init_hook_register(Arc::new(move |_s| {
                inner.fetch_add(1, Ordering::SeqCst);
            }));
        }));

        // First announce: outer hook runs, registers inner. Inner is
        // NOT called for this state.
        init_hook_announce(InitHookState::AtIocBuild);
        assert_eq!(inner_calls.load(Ordering::SeqCst), 0);

        // Second announce: both outer and the inner(s) run.
        init_hook_announce(InitHookState::AfterIocRunning);
        assert!(inner_calls.load(Ordering::SeqCst) >= 1);

        init_hooks::init_hook_free();
    }

    /// H1: state name strings match C `initHookName()`.
    #[test]
    fn init_hook_state_names_match_c() {
        assert_eq!(InitHookState::AtIocBuild.name(), "initHookAtIocBuild");
        assert_eq!(
            InitHookState::AfterInitDevSup.name(),
            "initHookAfterInitDevSup"
        );
        assert_eq!(
            InitHookState::AfterInitDatabase.name(),
            "initHookAfterInitDatabase"
        );
        assert_eq!(
            InitHookState::AfterIocRunning.name(),
            "initHookAfterIocRunning"
        );
    }

    #[tokio::test]
    async fn test_ioc_application_empty() {
        // An empty IocApplication with no script or records should start and stop cleanly
        // We can't easily test run() because it blocks on REPL, so test the wiring functions
        let db = Arc::new(PvDatabase::new());
        let factories = HashMap::new();
        let count = wire_device_support(&db, &factories, &None).await.unwrap();
        assert_eq!(count, 0);
    }

    #[tokio::test]
    async fn test_wire_device_support_no_dtyp() {
        use crate::server::records::ai::AiRecord;

        let db = Arc::new(PvDatabase::new());
        db.add_record("TEST", Box::new(AiRecord::new(0.0)))
            .await
            .unwrap();

        let factories = HashMap::new();
        let count = wire_device_support(&db, &factories, &None).await.unwrap();
        assert_eq!(count, 0); // No DTYP set, so no wiring
    }

    /// Round-8 regression: `wire_device_support` (the IocApplication
    /// startup-script device-support attach path) MUST forward
    /// info(...) tags to the driver via `apply_record_info`. Round-6
    /// only patched the IocBuilder path; without this fix, IOCs
    /// loaded entirely through iocsh `dbLoadRecords` lose every
    /// `info()` tag the driver depends on (e.g. asyn `asyn:READBACK`).
    #[tokio::test]
    async fn wire_device_support_forwards_info_tags_to_driver() {
        use crate::server::device_support::{DeviceReadOutcome, DeviceSupport};
        use crate::server::record::ScanType;
        use crate::server::records::ai::AiRecord;
        use std::sync::{Arc as StdArc, Mutex as StdMutex};

        // Recording device support — captures the info map it received
        // via apply_record_info so the test can assert on its contents.
        struct RecordingDev {
            seen: StdArc<StdMutex<HashMap<String, String>>>,
        }
        impl DeviceSupport for RecordingDev {
            fn write(&mut self, _record: &mut dyn crate::server::record::Record) -> CaResult<()> {
                Ok(())
            }
            fn dtyp(&self) -> &str {
                "TestRecording"
            }
            fn read(
                &mut self,
                _record: &mut dyn crate::server::record::Record,
            ) -> CaResult<DeviceReadOutcome> {
                Ok(DeviceReadOutcome::ok())
            }
            fn apply_record_info(&mut self, info: &HashMap<String, String>) {
                let mut g = self.seen.lock().unwrap();
                *g = info.clone();
            }
            fn set_record_info(&mut self, _name: &str, _scan: ScanType) {}
        }

        let seen = StdArc::new(StdMutex::new(HashMap::<String, String>::new()));
        let seen_factory = seen.clone();
        let mut factories: HashMap<String, DeviceSupportFactory> = HashMap::new();
        factories.insert(
            "TestRecording".to_string(),
            Box::new(move || {
                Box::new(RecordingDev {
                    seen: seen_factory.clone(),
                })
            }),
        );

        let db = Arc::new(PvDatabase::new());
        db.add_record("AI:WITH:INFO", Box::new(AiRecord::new(0.0)))
            .await
            .unwrap();
        // Populate the record's info map — exactly what
        // IocBuilder/iocsh now do after loading info(...) directives.
        let rec = db.get_record("AI:WITH:INFO").await.unwrap();
        {
            let mut inst = rec.write().await;
            inst.common.dtyp = "TestRecording".to_string();
            inst.set_info("asyn:READBACK", "1");
            inst.set_info("Q:group", "demo");
        }

        let count = wire_device_support(&db, &factories, &None).await.unwrap();
        assert_eq!(count, 1, "device support must have attached");

        // The recording driver should have observed both tags via
        // apply_record_info — proves the round-6 hook fires from the
        // IocApplication batch-wiring path too.
        let observed = seen.lock().unwrap().clone();
        assert_eq!(observed.get("asyn:READBACK").map(String::as_str), Some("1"));
        assert_eq!(observed.get("Q:group").map(String::as_str), Some("demo"));
    }
}