//! DAG shape builders — every operation as a template over the shared node
//! primitives. Pure (no I/O); each node carries its own `agent` (there is no
//! DAG-level agent) and **declares its own resources** with `.needs(…)` right
//! where it is constructed, rather than derived from its kind. Deriving made the
//! requirement a property of the *kind*, so a kind running under an ancestor
//! that already held the resource could get away with declaring nothing.
//!
//! **The one sanctioned exception is a brace** — a pure-resource-holder root
//! ([`NodeKind::AgentWindow`], [`NodeKind::DeployWindow`]) declaring for a
//! coordinated subtree whose members then declare nothing. See
//! `docs/coordinator.md`, _Braces_ — which also carries the per-operation DAG
//! shapes, so they are not restated here.
//!
//! Nodes are **named, not counted** — a template holds the handle
//! [`JobBuilder::node`] hands back, so an edge says which node it waits on. Why that
//! removes submit-time cycle validation: `docs/coordinator.md`.
//!
//! The hive-wide **power ops** (`stop` / `start` / `restart`) are NOT here:
//! their per-agent shape depends on live running state (an async
//! `lifecycle::is_running` read), so [`super::power`] assembles them out of
//! the primitives this module exports ([`rebuild_nodes`]).

use hive_jobq::TerminalState;

use super::model::{NodeKind, PermPayload};
use super::resource::Resource;
use super::{Handle, JobBuilder};

/// The `Rebuilt`-reporting tail pair for a rebuild-shaped DAG: the success node
/// gated on every group-root in `roots`, and the failure node gated on *its*
/// elimination.
///
/// Exactly one runs on a DAG that executed, and neither runs on one the operator
/// dropped — see [`hive_jobq::NodeRef::on_elimination_of`].
fn emit_rebuilt_tails(builder: &JobBuilder, agent: &str, roots: &[Handle<'_>]) {
    let ok = roots.iter().fold(
        builder.node(NodeKind::EmitRebuilt {
            agent: agent.to_owned(),
            ok: true,
        }),
        hive_jobq::NodeRef::after_ok,
    );
    // The failure branch needs *both*: the ok branch being ruled out (that is the
    // "something went wrong" signal) **and** every root actually finished. The
    // second half is easy to forget and gets the ordering wrong without it — a
    // failed `Prebuild` eliminates the ok branch immediately, while the recovery
    // `Reconcile` is still bringing the container back up, so reporting straight
    // off the elimination would announce the failure mid-recovery.
    let _failed = roots.iter().fold(
        builder
            .node(NodeKind::EmitRebuilt {
                agent: agent.to_owned(),
                ok: false,
            })
            .on_elimination_of(ok),
        hive_jobq::NodeRef::after_any,
    );
}

/// The approval-resolving tails for an approval-carrying DAG: one per outcome of
/// the DAG's single group-root `root`, each accepting only its own.
///
/// The `Cancelled` node is what keeps a dropped approval DAG from dangling its
/// row forever — its edge is the only one [`super::JobQueue::cancel`] spares.
fn resolve_approval_tails(builder: &JobBuilder, approval_id: i64, root: Handle<'_>) {
    for outcome in [
        TerminalState::Done,
        TerminalState::Failed,
        TerminalState::Cancelled,
    ] {
        let _ = builder
            .node(NodeKind::ResolveApproval {
                approval_id,
                outcome,
            })
            .on_outcome(root, &[outcome]);
    }
}

/// Declare one rebuild subgraph per agent onto the builder a running
/// [`NodeKind::MetaLock`] was handed.
///
/// **Into the emitter's own builder, not as new DAGs.** Growing in-DAG is what
/// roots each subgraph on the `MetaLock`, so the whole sweep (or meta-update
/// cascade) stays one unit of work the operator can watch and cancel, and every
/// rebuild builds against the lock the emitter just bumped.
///
/// Same reason as [`fanned_out_mechanical`] for living here: this was the
/// second construction site declaring nodes inline in an executor.
pub(crate) fn grown_rebuilds(builder: &JobBuilder, agents: &[String], relock: bool) {
    for agent in agents {
        rebuild_nodes(builder, agent, relock, None);
    }
}

/// As [`grown_rebuilds`], but each agent gets its `Signal` → `Drain` window
/// before being stopped. The boot sweep's flavour: it stops agents that were
/// mid-turn when the host came up, so they drain rather than being cut off.
pub(crate) fn grown_graceful_rebuilds(builder: &JobBuilder, agents: &[String], relock: bool) {
    for agent in agents {
        graceful_rebuild_nodes(builder, agent, relock, None);
    }
}

/// Declare the mechanical node a [`NodeKind::Reconcile`] planner fans out
/// (`Start` / `Stop`) onto the builder it was handed while running.
///
/// `Start` / `Stop` declare the agent lease they run under. Their `Reconcile`
/// parent is holding it already, so the declaration is a **re-entrant borrow**
/// — no second unit, no deadlock. It exists so the requirement belongs to the
/// node rather than to the fact that a `Reconcile` happens to fan it out.
///
/// Lives here rather than inline in `exec.rs` for the same reason every other
/// declaration does: this is the one construction site that was hiding in an
/// executor, which meant the only test of it had to re-declare the same two
/// calls itself and would have kept passing if the executor changed.
pub(crate) fn fanned_out_mechanical(builder: &JobBuilder, kind: NodeKind) {
    let lease = Resource::Agent(kind.agent().to_owned());
    let _ = builder.node(kind).needs(lease);
}

/// The graceful quiesce pair — `Signal` then `Drain`: ask the agent to run its
/// stop-checkpoint turn, then wait for it to go quiet. Returns the `Drain`
/// handle, which is what a caller edges its stop onto. The container stays
/// **up** throughout; the actual stop is the caller's next node.
///
/// `brace` must already hold [`Resource::Agent`] for its subtree; the pair
/// declares nothing and borrows it (see the module header). They are
/// dep-ordered **siblings**, not nested — nesting is only load-bearing where
/// `Signal` itself holds the lease, as in `submit.rs`'s `restart_chain`.
pub(crate) fn quiesce<'a>(builder: &'a JobBuilder, agent: &str, brace: Handle<'a>) -> Handle<'a> {
    let a = || agent.to_owned();
    let signal = builder.node(NodeKind::Signal { agent: a() }).part_of(brace);
    builder
        .node(NodeKind::Drain { agent: a() })
        .part_of(brace)
        .after_ok(signal)
}

/// The group-roots a [`rebuild_nodes`] subgraph exposes to its caller: what a
/// tail node edges onto, and what a follow-up node waits for.
///
/// Only the roots — a root's state *is* its subtree's roll-up, so these three
/// cover every node in the subgraph without the caller knowing its shape.
#[derive(Debug, Clone, Copy)]
pub(crate) struct RebuildRoots<'a> {
    /// The meta-repo preamble.
    pub meta_sync: Handle<'a>,
    /// The brace holding the agent lease and the build slot — its roll-up
    /// carries the whole mechanical subtree (`Prebuild`, the quiesce chain,
    /// `StopForUpdate` → `Swap` → `RebuildBookkeeping`).
    pub agent_window: Handle<'a>,
    /// The recovery/convergence tail root.
    pub reconcile: Handle<'a>,
}

impl<'a> RebuildRoots<'a> {
    /// The three roots as a slice, for edging a tail onto all of them.
    fn all(self) -> [Handle<'a>; 3] {
        [self.meta_sync, self.agent_window, self.reconcile]
    }
}

/// The rebuild node subtree — three group roots (`MetaSync`, the `AgentWindow`
/// brace, `Reconcile`). `after`, when given, is the node it chains behind. The
/// shape itself is in `docs/coordinator.md`; the code below is the source of
/// truth for it, so only the three choices a reader would otherwise undo are
/// called out here:
///
/// - **`MetaSync` is a sibling root, not the brace's parent** — it owns the
///   *global* `MetaWindow`, and a resource is held across the holder's whole
///   subtree, so parenting the rebuild under it would serialise every agent's
///   nix build behind one hive-wide window.
/// - **`StopForUpdate` waits on `Prebuild` as well as `Drain`** — running the
///   drain early is the win; taking the container *down* early is pure downtime.
/// - **`Reconcile` is a top-level root, not a child** — so it survives the
///   cancel-cascade of a failed brace and still converges the container
///   (recovery-start invariant). It takes a fresh lease; the gap is harmless
///   because it is idempotent.
fn rebuild_subtree<'a>(
    builder: &'a JobBuilder,
    agent: &str,
    relock: bool,
    graceful: bool,
    after: Option<Handle<'a>>,
) -> RebuildRoots<'a> {
    let a = || agent.to_owned();

    let mut meta_sync = builder
        .node(NodeKind::MetaSync { agent: a(), relock })
        .needs(Resource::MetaWindow);
    if let Some(after) = after {
        meta_sync = meta_sync.after_ok(after);
    }
    // The brace. Both resources are declared here, on one node, on purpose —
    // the queue acquires a node's resources atomically, so a single
    // multi-resource root can never hold one and block on another. Hoisting the
    // slot up costs nothing: a resource is held for the acquirer's whole
    // subtree, and the build slot already spanned the entire rebuild when
    // `Prebuild` was the one holding it.
    let agent_window = builder
        .node(NodeKind::AgentWindow { agent: a() })
        .needs(Resource::BuildSlot)
        .needs(Resource::Agent(a()))
        .after_ok(meta_sync);

    // Siblings under the brace: the build and the quiesce chain run
    // concurrently, borrowing the brace's grants rather than declaring their
    // own. Declaring the lease on both would make them mutually exclusive —
    // it is single-unit — which is exactly what this shape exists to avoid.
    let prebuild = builder
        .node(NodeKind::Prebuild { agent: a() })
        .part_of(agent_window);

    let drain = graceful.then(|| quiesce(builder, agent, agent_window));

    // The container goes down here, not earlier: `AfterOk` the build so a
    // failed build never stops a healthy container, and `AfterOk` the drain so
    // the agent has checkpointed.
    let mut stop_for_update = builder
        .node(NodeKind::StopForUpdate { agent: a() })
        .part_of(agent_window)
        .after_ok(prebuild);
    if let Some(drain) = drain {
        stop_for_update = stop_for_update.after_ok(drain);
    }

    let swap = builder
        .node(NodeKind::Swap { agent: a() })
        .part_of(stop_for_update);
    let _rebuild_bookkeeping = builder
        .node(NodeKind::RebuildBookkeeping { agent: a() })
        .part_of(stop_for_update)
        .after_ok(swap);

    let reconcile = builder
        .node(NodeKind::Reconcile { agent: a() })
        .needs(Resource::Agent(a()))
        .after_any(agent_window);

    RebuildRoots {
        meta_sync,
        agent_window,
        reconcile,
    }
}

/// The rebuild subtree, stopping the agent outright — the shape five of the six
/// call sites want. `relock` re-locks the meta flake inside `MetaSync`; `after`,
/// when given, is the node this subgraph chains behind. See
/// [`rebuild_subtree`] for the structure.
pub(crate) fn rebuild_nodes<'a>(
    builder: &'a JobBuilder,
    agent: &str,
    relock: bool,
    after: Option<Handle<'a>>,
) -> RebuildRoots<'a> {
    rebuild_subtree(builder, agent, relock, false, after)
}

/// As [`rebuild_nodes`], but the agent gets a `Signal` → `Drain` window to
/// finish the turn in flight before it is stopped. Costs up to one
/// `GRACEFUL_STOP_TIMEOUT` per subgraph, and those overlap across agents.
///
/// A separate entry point rather than a flag because `graceful` does not
/// *prepend* nodes — it **re-parents** the stop root, so a caller cannot
/// declare it without being handed the internals. Only the boot sweep wants it.
pub(crate) fn graceful_rebuild_nodes<'a>(
    builder: &'a JobBuilder,
    agent: &str,
    relock: bool,
    after: Option<Handle<'a>>,
) -> RebuildRoots<'a> {
    rebuild_subtree(builder, agent, relock, true, after)
}

/// The rebuild subgraph a [`NodeKind::DeployApply`] grows into its own DAG once
/// the merge has landed and `prepare_deploy` has staged the lock, plus the
/// [`NodeKind::FinalizeDeploy`] that closes the window behind it.
///
/// `relock = false` is the whole reason this composes: `prepare_deploy` already
/// relocked and staged `flake.lock`, so the appended `MetaSync` must do the dir
/// prep + `sync_agents` *without* re-locking over it.
///
/// `FinalizeDeploy` waits on **two** roots, which together reproduce the gate
/// the old fused node had around its inline `rebuild_no_meta` call:
/// - `AfterOk` `Prebuild` — a parent's state is its roll-up, so this is `Done`
///   only once `StopForUpdate` → `Swap` → `RebuildBookkeeping` all are (a failed *or*
///   cancelled child rolls the parent up `Failed`). That's the old
///   `build_result`.
/// - `AfterOk` `Reconcile` — the old call passed `deferred_start = false` on
///   purpose: the container had to come back up *before* the deploy was
///   finalized. `Reconcile` alone would not do, being `AfterAny` — it reaches
///   `Done` even after a failed `Swap`.
///
/// Declared into a **running** `DeployApply`'s own builder, not submitted: the
/// roots below become children of that node, which puts them inside the
/// `DeployWindow`'s subtree — so the `MetaWindow` this subgraph's `MetaSync`
/// and `FinalizeDeploy` declare is re-entered from the ancestor already holding
/// it rather than deadlocking against it.
pub(crate) fn deploy_rebuild_nodes(builder: &JobBuilder, agent: &str, approval_id: i64) {
    let roots = rebuild_nodes(builder, agent, false, None);
    let _finalize = builder
        .node(NodeKind::FinalizeDeploy {
            agent: agent.to_owned(),
            approval_id,
        })
        .needs(Resource::MetaWindow)
        .after_ok(roots.agent_window)
        .after_ok(roots.reconcile);
}

/// One uniform rebuild shape — no `was_running` branch. `StopForUpdate`
/// noops when already down; the tail `Reconcile` auto-noops the start
/// when `wanted = Offline` (a rebuild of a deliberately-stopped agent
/// leaves it stopped). `relock = false` only for meta-update cascade
/// children.
///
/// Closed by an [`NodeKind::EmitRebuilt`] tail edged onto all three group-roots
/// (`MetaSync`, the `AgentWindow` brace, `Reconcile`) — the brace's roll-up
/// carries the whole `StopForUpdate`→`Swap`→`RebuildBookkeeping` subtree, so
/// those three cover every node. Edging `Reconcile` alone would not do: it is
/// `AfterAny` the brace, so it reaches `Done` even after a failed swap and the
/// tail would report success.
///
/// Returns those three roots, so a caller that needs to wait on the rebuild can
/// name them.
pub fn rebuild(builder: &JobBuilder, agent: &str, relock: bool) -> Vec<hive_jobq::NodeGuid> {
    let roots = rebuild_nodes(builder, agent, relock, None);
    emit_rebuilt_tails(builder, agent, &roots.all());
    vec![
        roots.meta_sync.guid(),
        roots.agent_window.guid(),
        roots.reconcile.guid(),
    ]
}

/// Approval-driven deploy (`MergeConfigPr`) as a phase subtree rather than the
/// single opaque node it used to be. Structure:
/// - `DeployWindow` (**root**): the resource holder — global meta window, agent
///   lease, build slot — held across every child below. No work of its own; it
///   reaches `Finishing` immediately and the children run inside it.
/// - `MergeVerify` (child): drift-gate + fetch + eval-verify. Mutates nothing,
///   so a failure here cancel-cascades its siblings with the forge and the
///   applied repo exactly as they were.
/// - `DeployApply` (child, `AfterOk` `MergeVerify`): the irreversible half —
///   ff-merge + `prepare_deploy`. It doesn't rebuild inline; it grows
///   [`deploy_rebuild_nodes`] into this DAG as its own children, so the build
///   and the closing `FinalizeDeploy` are real nodes under the same window.
/// - `DeployTail` (child, `AfterAny` `DeployApply`): the compensation +
///   bookkeeping tail — rollback when a merge landed unfinalized, forge tag
///   mirror, PR failure comment (see [`NodeKind::DeployTail`]).
///
/// - `ResolveApproval` (**root**, `AfterAny` `DeployWindow`): resolves the
///   approval row. A root rather than another child, so it isn't inside the
///   window's resource subtree — it runs once the window has released the meta
///   window, lease and build slot. One edge suffices here: `DeployWindow` is the
///   DAG's only other group-root, so its roll-up already *is* the whole
///   pipeline's outcome.
///
/// The window still spans the container build, as it must: `prepare_deploy`
/// leaves `flake.lock` staged-uncommitted for the build's whole duration.
pub fn approval_deploy(builder: &JobBuilder, agent: &str, approval_id: i64) {
    let a = || agent.to_owned();
    // The window is the widest holder in the tree: it brackets a nix
    // build (`BuildSlot`), takes the container down across the swap
    // (`Agent`), and serialises the meta mutation its subtree performs
    // (`MetaWindow`). All three are held for its whole subtree, which
    // is what lets the appended rebuild's `MetaSync` and the
    // `FinalizeDeploy` re-enter rather than contend.
    let window = builder
        .node(NodeKind::DeployWindow {
            agent: a(),
            approval_id,
        })
        .needs(Resource::BuildSlot)
        .needs(Resource::Agent(a()))
        .needs(Resource::MetaWindow);
    let verify = builder
        .node(NodeKind::MergeVerify {
            agent: a(),
            approval_id,
        })
        .part_of(window);
    let apply = builder
        .node(NodeKind::DeployApply {
            agent: a(),
            approval_id,
        })
        .part_of(window)
        .after_ok(verify);
    let _tail = builder
        .node(NodeKind::DeployTail {
            agent: a(),
            approval_id,
        })
        .part_of(window)
        .after_any(apply);

    resolve_approval_tails(builder, approval_id, window);
}

/// First-deploy spawn (approval-driven): `Provision` (proposed/applied
/// repos, state subvolume, meta registration) then `Create`
/// (`nixos-container create`), drop-in write, then `Reconcile` starts
/// the container (`wanted = Up` written at approve time). All-or-nothing:
/// `Provision` (lease-exempt, precedes the container) is the group root;
/// `Create` (child) owns the agent lease; `WriteDropin` + `Reconcile`
/// (children of `Create`) borrow it. A failure cancel-cascades the rest —
/// unlike rebuild there's no recovery-reconcile (nothing to converge if the
/// container was never created). Closed by a `ResolveApproval` tail root edged
/// `AfterAny` onto `Provision` — the DAG's only other group-root, so its roll-up
/// already carries the whole cascade.
pub fn spawn(builder: &JobBuilder, agent: &str, approval_id: i64) {
    let a = || agent.to_owned();
    let provision = builder
        .node(NodeKind::Provision { agent: a() })
        .needs(Resource::MetaWindow);
    let create = builder
        .node(NodeKind::Create { agent: a() })
        .needs(Resource::BuildSlot)
        .needs(Resource::Agent(a()))
        .part_of(provision);
    let dropin = builder
        .node(NodeKind::WriteDropin { agent: a() })
        .needs(Resource::Agent(a()))
        .part_of(create);
    let _reconcile = builder
        .node(NodeKind::Reconcile { agent: a() })
        .needs(Resource::Agent(a()))
        .part_of(create)
        .after_ok(dropin);

    resolve_approval_tails(builder, approval_id, provision);
}

/// Perm change: commit the JSON file(s), then the rebuild subgraph so
/// the updated `HIVE_TOOL_GROUPS` / `HIVE_CAPABILITIES` env var takes
/// effect in the container. Group-roots are `WritePermFile` plus the rebuild
/// subgraph's `MetaSync` / `Prebuild` / `Reconcile`, so the `EmitRebuilt` tail
/// edges all four.
pub fn perm_change(builder: &JobBuilder, agent: &str, payload: PermPayload) {
    let write = builder
        .node(NodeKind::WritePermFile {
            agent: agent.to_owned(),
            payload,
        })
        .needs(Resource::MetaWindow);
    let roots = rebuild_nodes(builder, agent, true, Some(write));
    emit_rebuilt_tails(
        builder,
        agent,
        &[write, roots.meta_sync, roots.agent_window, roots.reconcile],
    );
}

/// Meta-input lock bump. The `MetaLock` executor grows one rebuild subgraph
/// per affected agent into *this same* DAG on completion (declared onto the
/// builder it was handed) — appended *after* the bump lands so their prebuilds
/// run against the post-bump lock, and a failed bump appends nothing
/// (replacing the old fan-out-child-DAGs dance).
/// `transient = Rebuilding` because those appended subgraphs are rebuilds:
/// it's applied per-agent at claim time (the `MetaLock` head needs no lease,
/// so the "hyperhive" pseudo-agent gets no pill), giving each cascade agent
/// crash-watch suppression during its `Swap` — the property the old child
/// `Rebuild` DAGs carried via their own transient.
pub fn meta_update(builder: &JobBuilder, inputs: Vec<String>, approval_id: Option<i64>) {
    let lock = builder
        .node(NodeKind::MetaLock {
            sweep: false,
            fanout: None,
            inputs,
        })
        .needs(Resource::BuildSlot)
        .needs(Resource::MetaWindow);
    // The bump itself has no side effect, so an operator-driven one ends
    // at the `MetaLock`; an approval-driven one still has its row to
    // resolve and gets the per-outcome tails edged onto that single
    // group-root — whose roll-up covers the rebuild subgraphs `MetaLock`
    // grows into itself.
    if let Some(approval_id) = approval_id {
        resolve_approval_tails(builder, approval_id, lock);
    }
}

/// Topology move(s) as a single-node DAG. `moves` is `(child, new_parent)`
/// pairs — len 1 for `set-parent`, len N for `set-parent-bulk`, applied
/// uniformly by the one [`NodeKind::Reparent`] node (which holds the global
/// meta window for its duration, same precedent as [`NodeKind::WritePermFile`]).
/// No rebuild subgraph: `topology.json` is read live by every consumer
/// (dashboard tree, `<parent>`/`<children>` sentinel routing, permission
/// checks), so a parent move needs no container rebuild to take effect.
/// No transient pill either — the node is agentless (no lease to hang one
/// off of) and near-instant. No tail node: the write is the whole effect.
///
/// Returns the single node's guid so a caller can wait on it.
pub fn reparent(
    builder: &JobBuilder,
    moves: Vec<(hive_types::Ident, Option<hive_types::Ident>)>,
) -> hive_jobq::NodeGuid {
    builder
        .node(NodeKind::Reparent { moves })
        .needs(Resource::MetaWindow)
        .guid()
}

// The boot is assembled inline in `workers/auto_update.rs::submit_boot_tree`
// as ONE `Boot` DAG (a sweep `MetaLock` root that grows rebuild subgraphs
// in-DAG, plus a `Reconcile` root per drifted agent) — no anchor node and no
// per-agent child DAGs.