//! Node executors — one async fn per [`NodeKind`], each a thin wrapper
//! over existing `lifecycle.rs` / `meta.rs` / `actions.rs` code. Node
//! executors keep their own internal error handling where it exists
//! today (cold-start fallback inside `Reconcile`, non-fatal boot-time
//! lock bump inside the sweep `MetaLock`, warn-only forge sync in the
//! `Swap` tail); DAG-level failure handling is cancel-downstream in
//! the queue.

use std::sync::Arc;

use anyhow::{Context as _, Result};

use super::model::{NodeKind, State, Template};
use super::{Claim, TerminalDag};
use crate::coordinator::Coordinator;
use crate::power::{ReconcileAction, reconcile_action};

/// Max time `Drain` waits for the harness to run its stop-checkpoint
/// turn before falling back to the hard stop. Generous — a checkpoint
/// turn can take a while — but bounded so a wedged agent never blocks
/// the stop indefinitely. Drains hold no build slot, so a whole-hive
/// graceful stop overlaps every agent's drain instead of serialising
/// N × this timeout.
pub const GRACEFUL_STOP_TIMEOUT: std::time::Duration = std::time::Duration::from_mins(3);

/// Extra signal an executor hands back to the scheduler alongside
/// success.
#[derive(Debug, Default)]
pub struct NodeOutput {
    /// Agents to fan child `Rebuild` DAGs out for (`MetaLock` only).
    pub fanout: Vec<String>,
}

/// Step-label + build-log sink for one claimed node.
struct Ctx<'a> {
    coord: &'a Arc<Coordinator>,
    dag_id: u64,
    node_id: super::NodeId,
}

impl Ctx<'_> {
    fn step(&self, step: &str) {
        if self
            .coord
            .job_queue
            .set_step(self.dag_id, self.node_id, step)
        {
            self.coord.emit_rebuild_queue_snapshot();
        }
    }

    fn build_log(&self, log_id: i64) {
        if self
            .coord
            .job_queue
            .set_build_log_id(self.dag_id, self.node_id, log_id)
        {
            self.coord.emit_rebuild_queue_snapshot();
        }
    }
}

/// Run one claimed node to completion. Called from a task the
/// scheduler spawns per claim; the `Result` (stringified) becomes the
/// node's terminal state.
pub(super) async fn run_node(coord: &Arc<Coordinator>, claim: &Claim) -> Result<NodeOutput> {
    let ctx = Ctx {
        coord,
        dag_id: claim.dag_id,
        node_id: claim.node_id,
    };
    match &claim.kind {
        NodeKind::Prebuild { relock } => run_prebuild(coord, claim, &ctx, *relock).await,
        NodeKind::Swap => run_swap(coord, claim, &ctx).await,
        NodeKind::Create => run_create(coord, claim, &ctx).await,
        NodeKind::MetaLock { sweep, fanout } => {
            run_meta_lock(coord, claim, &ctx, *sweep, fanout.clone()).await
        }
        NodeKind::Reconcile => run_reconcile(coord, claim, &ctx).await,
        NodeKind::StopForUpdate => run_stop_for_update(coord, claim, &ctx).await,
        NodeKind::Signal => Ok(run_signal(coord, claim, &ctx)),
        NodeKind::Drain => run_drain(coord, claim, &ctx).await,
        NodeKind::WriteDropin => run_write_dropin(coord, claim).await,
        NodeKind::WritePermFile => run_write_perm_file(coord, claim, &ctx).await,
        NodeKind::ApprovalDeploy => run_approval_deploy(coord, claim).await,
    }
}

/// Out-of-band toplevel build while the container keeps serving: meta
/// sync + optional per-agent relock, then warm
/// `system.build.toplevel` so the later `Swap` hits cache and skips
/// straight to the profile-swap.
async fn run_prebuild(
    coord: &Arc<Coordinator>,
    claim: &Claim,
    ctx: &Ctx<'_>,
    relock: bool,
) -> Result<NodeOutput> {
    let name = &claim.agent;
    // Prebuild runs while the agent is still up — the runtime dir and
    // MCP listener already exist. Use the pure path accessor; no need
    // to re-register the listener (event-driven: registered at start/create).
    let agent_dir = Coordinator::agent_dir(name);
    let hive = coord.hive_env();
    let paths = Coordinator::agent_paths(name, agent_dir);
    crate::lifecycle::prepare_rebuild_dirs(name, &paths).await?;
    // Idempotent meta sync so a manual rebuild can also recover from a
    // divergent meta repo; then bump just this agent's input. `relock =
    // false` only for meta-update cascade children, where re-locking
    // would revert the bump the cascade just committed. Both run under
    // the deploy-window gate so they can never land inside another
    // node's staged prepare→finalize window; the gate drops before the
    // (long) toplevel build, which only reads the store.
    {
        let _window = crate::meta::exclusive().await;
        let agents = crate::lifecycle::agents_for_meta_listing().await?;
        crate::meta::sync_agents(&hive, &agents).await?;
        if relock {
            crate::meta::lock_update_for_rebuild(name).await?;
        }
    }
    ctx.step("nix build");
    let flake_ref = format!("{}#{name}", crate::meta::meta_dir().display());
    crate::lifecycle::prebuild_toplevel(name, &flake_ref, &|log_id| ctx.build_log(log_id)).await?;
    Ok(NodeOutput::default())
}

/// Profile-swap: re-apply drop-ins (rebuild is the reconcile verb),
/// `nixos-container update`, then the post-rebuild bookkeeping tail
/// (rev marker, `Rebuilt` event, forge/matrix sync, kick, rescan).
/// The recovery-start on failure is NOT here — the DAG's tail
/// `Reconcile` runs after this node terminal ok *or* fail.
async fn run_swap(coord: &Arc<Coordinator>, claim: &Claim, ctx: &Ctx<'_>) -> Result<NodeOutput> {
    let name = &claim.agent;
    // Swap runs on an already-existing (stopped) container — runtime dir
    // and listener were created earlier. Pure path accessor suffices.
    let agent_dir = Coordinator::agent_dir(name);
    let hive = coord.hive_env();
    let paths = Coordinator::agent_paths(name, agent_dir);
    let result =
        crate::lifecycle::swap_update(name, &hive, &paths, &|step| ctx.step(step), &|log_id| {
            ctx.build_log(log_id);
        })
        .await;
    match &result {
        Ok(()) => {
            if let Some(rev) = crate::auto_update::current_flake_rev(&coord.hyperhive_flake)
                && let Err(e) = std::fs::write(crate::auto_update::rev_marker_path(name), rev)
            {
                tracing::warn!(%name, error = ?e, "write rev marker failed");
            }
            // The `Rebuilt` manager event fires exactly once per DAG
            // from the terminal hook — emitting ok here and letting a
            // failed tail `Reconcile` add a contradictory !ok would
            // double-report the same rebuild.
            ctx.step("forge sync");
            // Full forge + matrix sync on every successful rebuild so
            // the rebuild path is equivalent to the startup sweep:
            // tokens, config-repo mirror, meta access all recover
            // without a hive-c0re restart.
            crate::forge::sync_agent(name, crate::forge::core_token().as_deref()).await;
            crate::matrix::sync_agent_standalone(name).await;
            // Wake the agent on its next turn so claude sees a "you
            // were rebuilt" hint; rescan so dashboards drop the
            // "needs update" chip; lock bump → meta-inputs re-render.
            coord.kick_agent(name, "container rebuilt");
            coord.rescan_containers_and_emit().await;
            crate::dashboard::emit_meta_inputs_snapshot(coord);
        }
        Err(_) => {
            // The `Rebuilt { ok: false }` manager event fires once per
            // DAG from the terminal hook (any node may be the one that
            // failed); here only refresh the observed state.
            coord.rescan_containers_and_emit().await;
        }
    }
    result.map(|()| NodeOutput::default())
}

/// First-spawn provisioning + `nixos-container create` (atomic
/// build+create — no prebuild needed).
async fn run_create(coord: &Arc<Coordinator>, claim: &Claim, ctx: &Ctx<'_>) -> Result<NodeOutput> {
    let name = &claim.agent;
    let agent_dir = Coordinator::agent_dir(name);
    let hive = coord.hive_env();
    let paths = Coordinator::agent_paths(name, agent_dir);
    ctx.step("nixos-container create");
    // create_container registers the new agent in the meta flake
    // (sync_agents commit) before `nixos-container create` — hold the
    // deploy-window gate so that commit can't land inside another
    // node's staged deploy window.
    // Runtime dir creation and MCP listener registration are deferred to
    // the tail Reconcile (converge_start_preamble + register_agent) so this
    // node stays purely "provision + create", not "create + start".
    let _window = crate::meta::exclusive().await;
    crate::lifecycle::create_container(name, &hive, &paths).await?;
    Ok(NodeOutput::default())
}

/// Meta flake lock bump. Boot-sweep flavour is non-fatal (a failed
/// bump must not cancel the fan-out rebuilds — they proceed against
/// the current lock, exactly like today's sweep); the meta-update
/// flavour propagates errors, and a failed bump fans out nothing.
async fn run_meta_lock(
    coord: &Arc<Coordinator>,
    claim: &Claim,
    ctx: &Ctx<'_>,
    sweep: bool,
    fanout: Option<Vec<String>>,
) -> Result<NodeOutput> {
    if sweep {
        ctx.step("nix flake update hyperhive");
        let _window = crate::meta::exclusive().await;
        if let Err(e) = crate::meta::lock_update_hyperhive().await {
            tracing::warn!(error = ?e, "startup sweep: meta lock_update_hyperhive failed");
        }
        return Ok(NodeOutput {
            fanout: fanout.unwrap_or_default(),
        });
    }
    let _progress = coord.meta_update_guard();
    ctx.step("nix flake update");
    {
        let _window = crate::meta::exclusive().await;
        crate::meta::lock_update(&claim.inputs).await?;
    }
    // Lock file changed — meta-inputs panel re-renders.
    crate::dashboard::emit_meta_inputs_snapshot(coord);
    let cascade = match fanout {
        Some(list) => list,
        None => meta_update_cascade_agents(&claim.inputs).await,
    };
    Ok(NodeOutput { fanout: cascade })
}

/// Idempotent power converge: `wanted` (durable intent) vs observed.
async fn run_reconcile(
    coord: &Arc<Coordinator>,
    claim: &Claim,
    ctx: &Ctx<'_>,
) -> Result<NodeOutput> {
    let name = &claim.agent;
    let running = crate::lifecycle::is_running(name).await;
    let wanted = coord.power.get_or_seed(name, running)?;
    match reconcile_action(wanted, running) {
        ReconcileAction::Start => {
            // Node-local transient only when the DAG holds none (the
            // boot-reconcile template); lease-window guards otherwise
            // already cover this node.
            let _guard = claim
                .transient
                .is_none()
                .then(|| coord.transient_guard(name, crate::coordinator::TransientKind::Starting));
            // Run the typed start preamble: ensures the runtime dir
            // exists and writes the nspawn/resource-limits drop-ins.
            // The returned StartableAgent token is the only way to call
            // start_with_fallback — omitting this becomes a compile error.
            let agent_dir = Coordinator::agent_dir(name);
            let hive = coord.hive_env();
            let paths = Coordinator::agent_paths(name, agent_dir);
            let token = crate::lifecycle::converge_start_preamble(name, &hive, &paths).await?;
            ctx.step("nixos-container start");
            crate::lifecycle::start_with_fallback(token).await?;
            // Bind the MCP listener immediately after starting the container.
            // The preamble created the runtime dir; the container is now
            // coming up and will connect to this socket on its first turn.
            // Event-driven (no background poll) — c0re owns the listener
            // lifecycle, so register here rather than waiting for a sweep.
            coord.register_agent(name)?;
            coord.kick_agent(name, "container started");
            coord.rescan_containers_and_emit().await;
        }
        ReconcileAction::Stop => {
            let _guard = claim
                .transient
                .is_none()
                .then(|| coord.transient_guard(name, crate::coordinator::TransientKind::Stopping));
            ctx.step("nixos-container stop");
            crate::lifecycle::kill(name).await?;
            coord.unregister_agent(name);
            coord.notify_manager(&hive_sh4re::HelperEvent::Killed {
                agent: name.clone(),
            });
            coord.rescan_containers_and_emit().await;
        }
        ReconcileAction::Noop => {
            tracing::debug!(%name, wanted = wanted.as_str(), running, "reconcile: noop");
        }
    }
    Ok(NodeOutput::default())
}

/// Mechanical stop for the profile swap. Never *changes* `wanted`;
/// noop when already stopped.
async fn run_stop_for_update(
    coord: &Arc<Coordinator>,
    claim: &Claim,
    ctx: &Ctx<'_>,
) -> Result<NodeOutput> {
    let name = &claim.agent;
    if crate::lifecycle::is_running(name).await {
        // Seed a missing agent_power row from the PRE-stop observation
        // — the DAG's tail `Reconcile` observes only the mechanically
        // stopped state and would otherwise seed a running-but-unknown
        // agent as `Offline`, stranding it down after its own rebuild.
        if let Err(e) = coord.power.get_or_seed(name, true) {
            tracing::warn!(%name, error = ?e, "agent_power: pre-stop seed failed");
        }
        ctx.step("nixos-container stop");
        crate::lifecycle::kill(name).await?;
        coord.rescan_containers_and_emit().await;
    }
    Ok(NodeOutput::default())
}

/// Set the graceful fence + kick so the harness sees it promptly and
/// runs its one stop-checkpoint turn.
fn run_signal(coord: &Arc<Coordinator>, claim: &Claim, ctx: &Ctx<'_>) -> NodeOutput {
    ctx.step("graceful stop: signalling agent");
    coord.mark_graceful_stop(&claim.agent);
    coord.kick_agent(&claim.agent, "graceful stop requested");
    NodeOutput::default()
}

/// Await the harness clearing the fence (`GracefulStopComplete`) or
/// the timeout — either way the downstream `Reconcile` proceeds with
/// the actual stop.
async fn run_drain(coord: &Arc<Coordinator>, claim: &Claim, ctx: &Ctx<'_>) -> Result<NodeOutput> {
    let name = &claim.agent;
    ctx.step("graceful stop: draining");
    let deadline = std::time::Instant::now() + GRACEFUL_STOP_TIMEOUT;
    while coord.is_graceful_stop_pending(name) {
        if std::time::Instant::now() >= deadline {
            tracing::warn!(agent = %name, "graceful stop: drain timed out — hard-stopping");
            break;
        }
        tokio::time::sleep(std::time::Duration::from_millis(500)).await;
    }
    coord.clear_graceful_stop(name);
    Ok(NodeOutput::default())
}

/// `set_nspawn_flags` + `set_resource_limits` + daemon-reload.
async fn run_write_dropin(coord: &Arc<Coordinator>, claim: &Claim) -> Result<NodeOutput> {
    let name = &claim.agent;
    // write_dropins only needs the path value to build AgentPaths; the
    // dir doesn't need to exist at this point (created by ensure_runtime
    // on the upstream Prebuild/Start node).
    let agent_dir = Coordinator::agent_dir(name);
    let hive = coord.hive_env();
    let paths = Coordinator::agent_paths(name, agent_dir);
    crate::lifecycle::write_dropins(name, &hive, &paths).await?;
    Ok(NodeOutput::default())
}

/// Write + commit the perm file(s) (fused under `META_LOCK` so the
/// working tree is never left dirty), then emit the P3RM1SS10NS-tab
/// snapshots so the dashboard reflects the new assignment.
async fn run_write_perm_file(
    coord: &Arc<Coordinator>,
    claim: &Claim,
    ctx: &Ctx<'_>,
) -> Result<NodeOutput> {
    use super::model::PermPayload;
    let name = &claim.agent;
    ctx.step("writing + committing perm file");
    // Deploy-window gate: a perm commit landing inside another node's
    // staged prepare→finalize window would sweep the staged deploy
    // lock into its commit (the commits are also path-limited in
    // meta.rs — belt and braces).
    let _window = crate::meta::exclusive().await;
    match &claim.perm_payload {
        Some(PermPayload::ToolGroups { groups }) => {
            crate::meta::commit_tool_groups(name, groups)
                .await
                .with_context(|| format!("commit tool-groups for {name}"))?;
            coord.emit_tool_groups_snapshot();
        }
        Some(PermPayload::Capabilities { caps }) => {
            crate::meta::commit_capabilities(name, caps)
                .await
                .with_context(|| format!("commit capabilities for {name}"))?;
            coord.emit_capabilities_snapshot();
        }
        Some(PermPayload::Combined { groups, caps }) => {
            crate::meta::commit_perms(name, groups.as_deref(), caps.as_deref())
                .await
                .with_context(|| format!("commit perms for {name}"))?;
            if groups.is_some() {
                coord.emit_tool_groups_snapshot();
            }
            if caps.is_some() {
                coord.emit_capabilities_snapshot();
            }
        }
        None => anyhow::bail!(
            "perm_change dag {} for {name} is missing perm_payload",
            claim.dag_id
        ),
    }
    Ok(NodeOutput::default())
}

/// Opaque approval deploy pipeline: `ApplyCommit` and `MergeConfigPr`
/// both end in a container rebuild; branch on the approval row's kind
/// (the authoritative source). The two-phase prepare/finalize/abort
/// meta deploy — and the approval resolution — stay inside
/// `actions.rs` in v1 (design doc §9).
async fn run_approval_deploy(coord: &Arc<Coordinator>, claim: &Claim) -> Result<NodeOutput> {
    let approval_id = claim
        .approval_id
        .with_context(|| format!("approval_deploy dag {} has no approval_id", claim.dag_id))?;
    // Hold the deploy-window gate for the whole prepare→finalize span:
    // `prepare_deploy` stages `flake.lock` uncommitted for the entire
    // container build, and no other meta mutation may land inside that
    // window (it would sweep the staged lock and neuter `abort_deploy`).
    let _window = crate::meta::exclusive().await;
    let kind = coord
        .approvals
        .get(approval_id)
        .ok()
        .flatten()
        .map(|a| a.kind);
    let result = if kind == Some(hive_sh4re::ApprovalKind::MergeConfigPr) {
        crate::actions::run_approval_merge_config_pr(coord, Some(claim.dag_id), approval_id).await
    } else {
        crate::actions::run_approval_apply_commit(coord, Some(claim.dag_id), approval_id).await
    };
    result.map(|()| NodeOutput::default())
}

/// Terminal-roll-up hook, fired exactly once per DAG (node completion
/// and cancel paths alike — the queue buffers roll-ups and the
/// scheduler drains them). Three concerns:
/// - approval DAGs resolve their approval row (except the opaque
///   deploy pipeline, which resolves inside its node — unless it was
///   cancelled while still queued and the node never ran);
/// - non-approval rebuild-shaped DAGs emit exactly one `Rebuilt`
///   manager event: ok on `Done`, !ok on `Failed`, none on cancel;
/// - a cancelled power-op DAG reverts the `wanted` intent its submit
///   wrote: the operator's cancel means "don't do it", so intent
///   snaps back to the observed state instead of the flip executing
///   as a surprise side effect of some later reconcile.
pub(super) async fn on_dag_terminal(coord: &Arc<Coordinator>, terminal: &TerminalDag) {
    if terminal.state == State::Cancelled
        && matches!(
            terminal.template,
            Template::Start | Template::Stop | Template::GracefulStop | Template::Restart
        )
    {
        let running = crate::lifecycle::is_running(&terminal.agent).await;
        if let Err(e) = coord
            .power
            .set(&terminal.agent, crate::power::Wanted::from_running(running))
        {
            tracing::warn!(agent = %terminal.agent, error = ?e, "agent_power: cancel revert failed");
        }
    }
    if terminal.approval_id.is_some() {
        crate::actions::resolve_approval_dag(coord, terminal).await;
        return;
    }
    if matches!(terminal.template, Template::Rebuild | Template::PermChange) {
        match terminal.state {
            State::Done => coord.notify_manager(&hive_sh4re::HelperEvent::Rebuilt {
                agent: terminal.agent.clone(),
                ok: true,
                note: None,
                sha: None,
                tag: None,
            }),
            State::Failed => coord.notify_manager(&hive_sh4re::HelperEvent::Rebuilt {
                agent: terminal.agent.clone(),
                ok: false,
                note: terminal.error.clone(),
                sha: None,
                tag: None,
            }),
            _ => {}
        }
    }
}

/// Compute which agents a `nix flake update <inputs>` on the meta
/// flake affects — the fan-out set for `MetaUpdate` DAGs. Empty
/// `inputs` or any input under `hyperhive` → every container;
/// otherwise just the agents named by `agent-<name>` inputs.
/// Topology-sorted so parents rebuild before their children.
pub async fn meta_update_cascade_agents(inputs: &[String]) -> Vec<String> {
    let touched_hyperhive = inputs
        .iter()
        .any(|i| i == "hyperhive" || i.starts_with("hyperhive/"));
    let touched_agents: Vec<String> = inputs
        .iter()
        .filter_map(|i| i.strip_prefix("agent-"))
        .map(|rest| rest.split('/').next().unwrap_or(rest).to_owned())
        .collect();
    let mut names = if touched_hyperhive || inputs.is_empty() {
        crate::lifecycle::list()
            .await
            .unwrap_or_default()
            .into_iter()
            .filter_map(|c| {
                c.strip_prefix(crate::lifecycle::AGENT_PREFIX)
                    .map(str::to_owned)
            })
            .collect()
    } else {
        touched_agents
    };
    let topo = crate::topology::read();
    crate::auto_update::topology_sort(&mut names, &topo);
    names
}