nexo-core 0.1.19

//! Runtime config reload coordinator.
//!
//! Watches the config directory, re-runs boot validation, builds a
//! fresh `RuntimeSnapshot` per agent, and dispatches
//! `ReloadCommand::Apply` to each live runtime through the per-agent
//! mpsc channel the runtime exposed via `reload_sender()`. The
//! existing per-agent snapshot is left in place whenever validation
//! fails or a snapshot cannot be built — servicing never drops to a
//! broken config.
//!
//! Scope: hot-swap of **existing** agents only. Adding a brand-new
//! agent id or removing a running one requires spawn/teardown
//! plumbing that lives in `src/main.rs` today.

use std::path::PathBuf;
use std::sync::Arc;
use std::time::{Duration, Instant};

use arc_swap::ArcSwapOption;
use dashmap::DashMap;
use nexo_broker::{AnyBroker, BrokerHandle};
use nexo_config::AppConfig;
use nexo_llm::LlmRegistry;
use tokio::sync::{mpsc, Mutex};
use tokio_util::sync::CancellationToken;

use crate::agent::runtime::ReloadCommand;
use crate::agent::spawn::{AgentSpawnerFn, SharedRuntimeContext};
use crate::runtime_snapshot::RuntimeSnapshot;
use crate::telemetry;

/// Per-agent state the coordinator needs to dispatch a reload. Held
/// in a `DashMap<String, AgentReloadHandle>` keyed by agent id.
///
/// `known_tools` captures the agent's tool surface at boot time —
/// builtins + plugins + MCP + extensions + skills, after the per-agent
/// allowlist prune. The coordinator uses it during reload so a typo
/// in `allowed_tools` (binding-level) fails the swap instead of
/// silently degrading to "agent has no tools at runtime".
pub struct AgentReloadHandle {
    pub reload_tx: mpsc::Sender<ReloadCommand>,
    pub known_tools: Arc<Vec<String>>,
}

/// Hook the reload coordinator runs after every successful swap.
/// Used to invalidate process-wide caches that hold a stale view of
/// data the reload may have changed (e.g. `PairingGate`'s in-memory
/// allowlist cache, which would otherwise keep blocking a sender the
/// operator just `nexo pair seed`-ed). Hooks are best-effort; they
/// run sequentially under the same gate as the reload itself, so
/// keep them cheap (one mutex / one dashmap clear).
pub type PostReloadHook = Box<dyn Fn() + Send + Sync>;

/// Reload coordinator. One instance per process; `start` spawns the
/// file watcher and the broker `control.reload` subscriber.
pub struct ConfigReloadCoordinator {
    config_dir: PathBuf,
    runtimes: DashMap<String, AgentReloadHandle>,
    llm_registry: Arc<LlmRegistry>,
    version: Mutex<u64>,
    /// Serial gate so two overlapping triggers (watcher + CLI) don't
    /// race the snapshot build. The second trigger queues behind the
    /// first.
    gate: Mutex<()>,
    /// Broker handle attached at `start()`. `Some` once the daemon is
    /// up; the file-watcher branch uses it to publish
    /// `events.runtime.config.reloaded` after every successful swap so
    /// extensions and dashboards can react without polling.
    broker: ArcSwapOption<AnyBroker>,
    /// Cache-flush hooks fired after every successful reload. Locked
    /// by the same gate as the reload to keep the contract simple
    /// (no observer can run mid-swap).
    post_hooks: Mutex<Vec<PostReloadHook>>,
    /// Phase 81.32 — shared runtime context the coordinator
    /// hands to `spawn_agent_runtime` when an agent id appears
    /// in the new config that wasn't there before. `None` keeps
    /// the legacy "adding a new agent at runtime is not
    /// supported" rejection so tests that haven't wired the
    /// context stay on the old behaviour.
    shared_ctx: ArcSwapOption<SharedRuntimeContext>,
    /// Phase 81.32 c6 — spawner closure the coordinator invokes
    /// when an unknown agent id appears in `agents.yaml`. `None`
    /// keeps the legacy rejection ("adding a new agent at
    /// runtime is not supported"). Installed at boot via
    /// [`Self::set_spawner`] once `src/main.rs` has finished
    /// constructing all per-agent dependencies.
    spawner: ArcSwapOption<AgentSpawnerFn>,
    shutdown: CancellationToken,
}

/// Result returned by [`ConfigReloadCoordinator::reload`].
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub struct ReloadOutcome {
    pub version: u64,
    pub applied: Vec<String>,
    pub rejected: Vec<ReloadRejection>,
    pub elapsed_ms: u64,
}

#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub struct ReloadRejection {
    pub agent_id: Option<String>,
    pub reason: String,
}

impl ConfigReloadCoordinator {
    pub fn new(
        config_dir: PathBuf,
        llm_registry: Arc<LlmRegistry>,
        shutdown: CancellationToken,
    ) -> Self {
        Self {
            config_dir,
            runtimes: DashMap::new(),
            llm_registry,
            version: Mutex::new(0),
            gate: Mutex::new(()),
            broker: ArcSwapOption::from(None),
            post_hooks: Mutex::new(Vec::new()),
            shared_ctx: ArcSwapOption::from(None),
            spawner: ArcSwapOption::from(None),
            shutdown,
        }
    }

    /// Phase 81.32 — install the [`SharedRuntimeContext`] the
    /// reload coordinator hands to `spawn_agent_runtime` when an
    /// agent id appears in the freshly-loaded config that wasn't
    /// in the previous one. Without this wired, the legacy
    /// rejection ("adding a new agent at runtime is not
    /// supported") fires for unknown ids.
    ///
    /// Late-bindable so `src/main.rs` can build the coordinator
    /// before the boot-loop singletons are fully assembled and
    /// upgrade it once they are.
    pub fn with_shared_context(self, shared: Arc<SharedRuntimeContext>) -> Self {
        self.shared_ctx.store(Some(shared));
        self
    }

    /// Phase 81.32 — read-only handle to the configured shared
    /// context. `None` when [`Self::with_shared_context`] has
    /// not been called yet (legacy / test path).
    pub fn shared_context(&self) -> Option<Arc<SharedRuntimeContext>> {
        self.shared_ctx.load_full()
    }

    /// Phase 81.32 c6 — install the spawner closure invoked when
    /// an unknown agent id appears in `agents.yaml`. Late-bindable
    /// (same shape as [`Self::with_shared_context`]) so the
    /// coordinator can exist before the boot loop captures every
    /// per-agent dependency.
    pub fn set_spawner(&self, spawner: Arc<AgentSpawnerFn>) {
        self.spawner.store(Some(spawner));
    }

    /// Phase 81.32 c6 — read-only handle to the configured
    /// spawner. `None` when [`Self::set_spawner`] has not yet
    /// fired; callers fall back to the legacy "not supported"
    /// rejection.
    pub fn spawner(&self) -> Option<Arc<AgentSpawnerFn>> {
        self.spawner.load_full()
    }

    /// Phase 81.32 — uninstall the per-agent reload handle when
    /// an agent is hot-removed from `agents.yaml`. Returns the
    /// handle so the coordinator can drive its
    /// `ReloadCommand::Shutdown` send before dropping it.
    pub fn unregister(&self, agent_id: &str) -> Option<AgentReloadHandle> {
        self.runtimes.remove(agent_id).map(|(_, handle)| handle)
    }

    /// Register a closure that fires after every successful reload.
    /// Callers should add their cache-invalidation entry point here
    /// at boot. Hooks run inside the reload gate so they observe the
    /// new config but cannot themselves overlap a future reload.
    pub async fn register_post_hook(&self, hook: PostReloadHook) {
        self.post_hooks.lock().await.push(hook);
    }

    /// Register a live agent runtime's reload channel. Called once per
    /// agent during boot; subsequent reloads look these up to dispatch.
    pub fn register(
        &self,
        agent_id: impl Into<String>,
        reload_tx: mpsc::Sender<ReloadCommand>,
        known_tools: Arc<Vec<String>>,
    ) {
        self.runtimes.insert(
            agent_id.into(),
            AgentReloadHandle {
                reload_tx,
                known_tools,
            },
        );
    }

    /// Re-read config, validate, build snapshots, dispatch. Returns an
    /// aggregate outcome — successful ids in `applied`, per-agent or
    /// top-level failures in `rejected`. The coordinator never panics
    /// on a bad config; it logs + bumps the rejected counter and keeps
    /// old snapshots serving.
    pub async fn reload(&self) -> ReloadOutcome {
        let _gate = self.gate.lock().await;
        let started = Instant::now();
        let mut applied: Vec<String> = Vec::new();
        let mut rejected: Vec<ReloadRejection> = Vec::new();

        // 1. Load + env-resolve.
        let cfg = match AppConfig::load(&self.config_dir) {
            Ok(c) => c,
            Err(e) => {
                telemetry::inc_config_reload_rejected();
                tracing::warn!(error = %e, "config reload: load failed, keeping previous snapshot");
                rejected.push(ReloadRejection {
                    agent_id: None,
                    reason: format!("AppConfig::load: {e}"),
                });
                let current = *self.version.lock().await;
                return ReloadOutcome {
                    version: current,
                    applied,
                    rejected,
                    elapsed_ms: started.elapsed().as_millis() as u64,
                };
            }
        };

        // 2. Structural + provider validation (aggregate errors).
        // Providers are LLM yaml instance ids
        // (`anthropic-a5b8`), NOT factory ids — agents bind to
        // yaml-key instances that map to a factory via
        // `factory_type`. Mirror the boot validation path in
        // `src/main.rs::validate_agents_with_providers`.
        let known_providers =
            crate::agent::KnownProviders::new(cfg.llm.providers.keys().map(String::as_str));
        if let Err(e) = crate::agent::validate_agents_with_providers(
            &cfg.agents.agents,
            &cfg.plugins,
            &crate::agent::KnownTools::default(),
            &known_providers,
        ) {
            telemetry::inc_config_reload_rejected();
            tracing::warn!(error = %e, "config reload: validation failed, keeping previous snapshot");
            rejected.push(ReloadRejection {
                agent_id: None,
                reason: format!("validation: {e}"),
            });
            let current = *self.version.lock().await;
            return ReloadOutcome {
                version: current,
                applied,
                rejected,
                elapsed_ms: started.elapsed().as_millis() as u64,
            };
        }

        // 3. Bump version + build snapshots per agent present in BOTH
        //    old (registered handles) and new (config). Agents that
        //    disappear or appear are out of scope — we skip them
        //    with a rejection entry so the operator sees the diff.
        let mut version_guard = self.version.lock().await;
        *version_guard += 1;
        let new_version = *version_guard;
        drop(version_guard);

        for agent_cfg in &cfg.agents.agents {
            // Phase 81.32 c8 — unknown agent id (wizard create
            // path). Try the installed spawner first; fall back to
            // the legacy rejection only when no spawner was wired
            // (test harnesses / minimal embeddings). Closure
            // invocation drops the lock guard via `self.runtimes
            // .get(...)` returning `Some` only when an entry
            // already exists; the spawner branch runs OUTSIDE the
            // guard scope to keep `register(...)` reentrant.
            if !self.runtimes.contains_key(&agent_cfg.id) {
                let Some(spawner) = self.spawner() else {
                    rejected.push(ReloadRejection {
                        agent_id: Some(agent_cfg.id.clone()),
                        reason: "adding a new agent at runtime is not supported; \
                                 set a spawner via ConfigReloadCoordinator::set_spawner"
                            .into(),
                    });
                    continue;
                };
                match spawner.call(agent_cfg.clone()).await {
                    Ok(spawned) => {
                        self.register(spawned.agent_id.clone(), spawned.reload_tx, spawned.known_tools);
                        applied.push(spawned.agent_id.clone());
                        // Best-effort firehose notification — operators
                        // tail the broker events stream to see when a
                        // wizard-created agent goes live.
                        if let Some(b) = self.broker.load_full() {
                            let evt = nexo_broker::Event::new(
                                "events.runtime.agent.spawned",
                                "config_reload",
                                serde_json::json!({
                                    "agent_id": spawned.agent_id,
                                    "version": new_version,
                                }),
                            );
                            let _ = b.publish("events.runtime.agent.spawned", evt).await;
                        }
                        tracing::info!(
                            agent = %agent_cfg.id,
                            "hot-spawned agent via ConfigReloadCoordinator",
                        );
                    }
                    Err(e) => {
                        rejected.push(ReloadRejection {
                            agent_id: Some(agent_cfg.id.clone()),
                            reason: format!("spawn: {e}"),
                        });
                    }
                }
                continue;
            }
            let Some(handle) = self.runtimes.get(&agent_cfg.id) else {
                // Race window between `contains_key` + `get` —
                // operator hot-removed the agent mid-reload. Treat
                // as a rejection for this cycle; the next reload
                // re-detects.
                rejected.push(ReloadRejection {
                    agent_id: Some(agent_cfg.id.clone()),
                    reason: "agent vanished between hot-spawn check and snapshot build".into(),
                });
                continue;
            };

            // Per-agent post-assembly tool-name validation. Mirrors
            // the boot-path second-pass check so a binding's typo'd
            // `allowed_tools` rejects the reload instead of silently
            // landing a config that the runtime then has to translate
            // into a "tool not available" error every turn.
            let known_strs: Vec<&str> = handle.known_tools.iter().map(|s| s.as_str()).collect();
            let catalog = crate::agent::KnownTools::new(known_strs);
            if let Err(e) = crate::agent::validate_agent(agent_cfg, &cfg.plugins, &catalog)
            {
                rejected.push(ReloadRejection {
                    agent_id: Some(agent_cfg.id.clone()),
                    reason: format!("post-assembly validation: {e}"),
                });
                continue;
            }

            let snap = match RuntimeSnapshot::build(
                Arc::new(agent_cfg.clone()),
                &self.llm_registry,
                &cfg.llm,
                new_version,
            ) {
                Ok(s) => Arc::new(s),
                Err(e) => {
                    rejected.push(ReloadRejection {
                        agent_id: Some(agent_cfg.id.clone()),
                        reason: format!("snapshot build: {e}"),
                    });
                    continue;
                }
            };

            match handle.reload_tx.send(ReloadCommand::Apply(snap)).await {
                Ok(()) => applied.push(agent_cfg.id.clone()),
                Err(e) => rejected.push(ReloadRejection {
                    agent_id: Some(agent_cfg.id.clone()),
                    reason: format!("dispatch: {e}"),
                }),
            }
        }

        // 4. Detect removed agents (registered but absent from new cfg).
        // Phase 81.32 c9 — hot-teardown. Send `ReloadCommand::Shutdown`
        // to the per-agent runtime so it drops broker subs +
        // heartbeat tasks cleanly, then `unregister` the handle so
        // future reloads can `hot-spawn` the same id without
        // colliding with stale state. Best-effort event publish so
        // operators watching the firehose see hot-remove distinctly
        // from a regular reload.
        //
        // Collect removed ids first (can't mutate the map while
        // iterating it; DashMap reentrant remove panics on the same
        // shard).
        let removed_ids: Vec<String> = self
            .runtimes
            .iter()
            .filter_map(|entry| {
                let id = entry.key();
                if !cfg.agents.agents.iter().any(|a| &a.id == id) {
                    Some(id.clone())
                } else {
                    None
                }
            })
            .collect();
        for id in removed_ids {
            let Some(handle) = self.unregister(&id) else {
                continue;
            };
            // Best-effort `Shutdown` dispatch. A full mailbox /
            // closed channel here means the runtime task already
            // exited — log + carry on so the operator still sees
            // the agent disappear from `runtimes`.
            if let Err(e) = handle.reload_tx.send(ReloadCommand::Shutdown).await {
                tracing::warn!(
                    agent = %id,
                    error = %e,
                    "hot-remove: runtime mailbox dispatch failed (task may have exited already)",
                );
            }
            if let Some(b) = self.broker.load_full() {
                let evt = nexo_broker::Event::new(
                    "events.runtime.agent.removed",
                    "config_reload",
                    serde_json::json!({
                        "agent_id": id,
                        "version": new_version,
                    }),
                );
                let _ = b.publish("events.runtime.agent.removed", evt).await;
            }
            applied.push(id.clone());
            tracing::info!(agent = %id, "hot-removed agent via ConfigReloadCoordinator");
        }

        let elapsed_ms = started.elapsed().as_millis() as u64;
        telemetry::observe_config_reload_latency_ms(elapsed_ms);

        if !applied.is_empty() {
            telemetry::inc_config_reload_applied();
            tracing::info!(
                version = new_version,
                applied = ?applied,
                rejected_count = rejected.len(),
                elapsed_ms,
                "config reload applied",
            );
            // Run cache-invalidation hooks (e.g. PairingGate flush)
            // before publishing the reload event, so consumers see
            // the new state cleanly. We hold the lock briefly — the
            // gate above already prevents overlapping reloads, the
            // post-hooks lock just guards the registration list.
            let hooks = self.post_hooks.lock().await;
            for hook in hooks.iter() {
                hook();
            }
            drop(hooks);
            // Broadcast the event so extensions
            // and dashboards can react without polling. Non-fatal if
            // the publish fails; the metrics + log already record the
            // swap.
            if let Some(broker) = self.broker.load_full() {
                let payload = serde_json::json!({
                    "version": new_version,
                    "applied": &applied,
                    "rejected": &rejected,
                    "elapsed_ms": elapsed_ms,
                });
                let topic = "events.runtime.config.reloaded";
                let evt = nexo_broker::Event::new(topic, "config-reload", payload);
                if let Err(e) = broker.publish(topic, evt).await {
                    tracing::warn!(error = %e, "failed to publish events.runtime.config.reloaded");
                }
            }
        }
        if !rejected.is_empty() {
            tracing::warn!(
                version = new_version,
                rejected = ?rejected,
                "config reload: partial rejects",
            );
        }

        ReloadOutcome {
            version: new_version,
            applied,
            rejected,
            elapsed_ms,
        }
    }

    /// Start the watcher + broker subscriber. Returns immediately; the
    /// work runs on spawned tasks that honour `self.shutdown`.
    pub async fn start(
        self: Arc<Self>,
        broker: AnyBroker,
        reload: nexo_config::RuntimeReloadConfig,
    ) -> anyhow::Result<()> {
        if !reload.enabled {
            tracing::info!("config hot-reload disabled via runtime.yaml");
            return Ok(());
        }

        // Stash the broker so reload() can emit
        // `events.runtime.config.reloaded` regardless of whether the
        // trigger came from the file watcher or the CLI.
        self.broker.store(Some(Arc::new(broker.clone())));

        // File watcher → debounced notifications.
        let watcher_rx = crate::config_watch::spawn_config_watcher(
            self.config_dir.clone(),
            reload.extra_watch_paths.clone(),
            Duration::from_millis(reload.debounce_ms),
            self.shutdown.clone(),
        )?;
        let coord_watcher = Arc::clone(&self);
        tokio::spawn(async move {
            let mut rx = watcher_rx;
            while let Some(()) = rx.recv().await {
                if coord_watcher.shutdown.is_cancelled() {
                    break;
                }
                let _ = coord_watcher.reload().await;
            }
        });

        // Broker subscriber — manual triggers from `agent reload`.
        let mut sub = broker.subscribe("control.reload").await?;
        let coord_broker = Arc::clone(&self);
        let broker_clone = broker.clone();
        tokio::spawn(async move {
            loop {
                if coord_broker.shutdown.is_cancelled() {
                    break;
                }
                let Some(_event) = sub.next().await else {
                    break;
                };
                let outcome = coord_broker.reload().await;
                let ack_topic = "control.reload.ack";
                let payload = serde_json::to_value(&outcome)
                    .unwrap_or_else(|e| serde_json::json!({ "error": e.to_string() }));
                let evt = nexo_broker::Event::new(ack_topic, "config-reload", payload);
                if let Err(e) = broker_clone.publish(ack_topic, evt).await {
                    tracing::warn!(error = %e, "failed to publish control.reload.ack");
                }
            }
        });

        Ok(())
    }

    /// Current monotonic version (for telemetry / tests).
    pub async fn version(&self) -> u64 {
        *self.version.lock().await
    }
}

#[cfg(test)]
impl ConfigReloadCoordinator {
    /// Test-only: count of registered post hooks.
    /// Used by `register_plugin_registry_reload_hook` to verify
    /// it pushes exactly one hook.
    pub async fn post_hooks_len_for_test(&self) -> usize {
        self.post_hooks.lock().await.len()
    }

    /// Test-only: fire every registered post-hook in FIFO order.
    /// Mirrors the production fire path but skips the gate + reload
    /// itself; callers exercise the hook contract, not the reload
    /// mechanics.
    pub async fn fire_post_hooks_for_test(&self) {
        let hooks = self.post_hooks.lock().await;
        for hook in hooks.iter() {
            hook();
        }
    }
}

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

    #[tokio::test]
    async fn reload_with_no_config_dir_falls_back_to_defaults() {
        // Phase 93 — `AppConfig::load` tolerates a missing config dir
        // by returning `Default::default()` (same as the daemon's
        // zero-config boot path). A hot-reload pointed at a
        // nonexistent dir is therefore a clean no-op: defaults
        // validate, there are 0 agents to hot-swap, nothing rejected.
        let coord = Arc::new(ConfigReloadCoordinator::new(
            PathBuf::from("/nonexistent-config-dir-xyz"),
            Arc::new(LlmRegistry::with_builtins()),
            CancellationToken::new(),
        ));
        let outcome = coord.reload().await;
        assert!(
            outcome.rejected.is_empty(),
            "a missing config dir is tolerated, not rejected: {:?}",
            outcome.rejected
        );
        assert!(
            outcome.applied.is_empty(),
            "default config has no agents to apply: {:?}",
            outcome.applied
        );
    }

    #[tokio::test]
    async fn version_starts_at_zero() {
        let coord = ConfigReloadCoordinator::new(
            PathBuf::from("."),
            Arc::new(LlmRegistry::with_builtins()),
            CancellationToken::new(),
        );
        assert_eq!(coord.version().await, 0);
    }

    #[tokio::test]
    async fn post_hooks_register_and_can_be_invoked_in_order() {
        // Verify the hook list grows and runs in
        // registration order. The reload() success path that fires
        // them needs a full AppConfig on disk; that's covered by the
        // boot smoke tests. Here we just check the storage / FIFO.
        use std::sync::atomic::{AtomicUsize, Ordering};
        let coord = ConfigReloadCoordinator::new(
            PathBuf::from("."),
            Arc::new(LlmRegistry::with_builtins()),
            CancellationToken::new(),
        );
        let order = Arc::new(AtomicUsize::new(0));
        let a_witness = Arc::new(AtomicUsize::new(0));
        let b_witness = Arc::new(AtomicUsize::new(0));
        {
            let order = Arc::clone(&order);
            let w = Arc::clone(&a_witness);
            coord
                .register_post_hook(Box::new(move || {
                    w.store(order.fetch_add(1, Ordering::SeqCst) + 1, Ordering::SeqCst);
                }))
                .await;
        }
        {
            let order = Arc::clone(&order);
            let w = Arc::clone(&b_witness);
            coord
                .register_post_hook(Box::new(move || {
                    w.store(order.fetch_add(1, Ordering::SeqCst) + 1, Ordering::SeqCst);
                }))
                .await;
        }
        let hooks = coord.post_hooks.lock().await;
        assert_eq!(hooks.len(), 2);
        for hook in hooks.iter() {
            hook();
        }
        drop(hooks);
        assert_eq!(a_witness.load(Ordering::SeqCst), 1);
        assert_eq!(b_witness.load(Ordering::SeqCst), 2);
    }
}