diff --git a/docs/coordinator.md b/docs/coordinator.md index b02ed1ed..554ae58b 100644 --- a/docs/coordinator.md +++ b/docs/coordinator.md @@ -108,17 +108,28 @@ Notable collapses: ### Desired-state (spec vs status) -Per-agent power *intent* — `wanted: Up | Offline` — is durable in -`/var/lib/hyperhive/db/agent_power.sqlite` (`hive-c0re/src/power.rs`). +Per-agent power *intent* — `wanted: Up | Offline` — is durable as the +`agent_power` table in the coordinator DB (`hive-c0re/src/power.rs`). `container_view` remains the observed *status*; `Reconcile` nodes converge the two. Setting `wanted` is never a queued node: the submit layer (`job_queue/submit.rs`) writes the row synchronously, then submits the DAG whose `Reconcile` reads the fresh value — rapid toggles are last-writer-wins. -Power toggles never commit to the meta repo. Direct (non-queued) power paths — -`hivectl stop/start`, the admin-socket kill, the MCP kill tool — write -`wanted` too, so reconciles never undo an operator's stop. Agents without a -row are seeded from observed state on first touch (running ⇒ `Up`); destroy -removes the row. +Power toggles never commit to the meta repo. Every operator power surface — +dashboard buttons, the MCP tools, and `hivectl stop/start/restart/kill` — +rides the queue through that submit layer, so intent, lease serialization, +and crash-watch suppression can't drift per surface; the only direct starts +left are the root-agent bootstrap and infra containers (no lease, no +harness). Cancelling a still-queued power DAG reverts `wanted` to the +observed state — a cancel means "don't do it", not "do it later". Agents +without a row are seeded from observed state on first touch (running ⇒ +`Up`); destroy removes the row. + +The admin-socket responses carry the submitted DAG ids; `hivectl` polls +`HostRequest::QueueDag` (~1s) and prints a progress line per DAG — roll-up +glyph, template, agent, node chain with the running node's step label — so +CLI verbs block until their jobs finish (`--no-wait` opts out; failures exit +non-zero). Fan-out children joining a polled parent show up in the same +loop. ### Scheduler semantics diff --git a/docs/tools/hivectl-cli.md b/docs/tools/hivectl-cli.md index 44aa1d57..6c3fe955 100644 --- a/docs/tools/hivectl-cli.md +++ b/docs/tools/hivectl-cli.md @@ -271,8 +271,8 @@ Agent container management. Requires the hive-c0re daemon to be running (connect ###### **Subcommands:** * `list` — Show all managed agents with their status (running / needs-login / needs-update) and technical state (deployed sha, parent, pending reminders). The host roster overview; reuses the dashboard's per-agent aggregation. Requires the daemon running -* `restart` — Stop and start a single agent container without rebuilding config. Useful for "kick the container" when the process is stuck or the container needs a clean restart without changing the NixOS config -* `restart-all` — Stop and restart ALL managed agent containers in sequence. Iterates the live container list and restarts each one. Any per-agent failure is reported at the end rather than stopping mid-run, so all containers get a restart attempt +* `restart` — Stop and start a single agent container without rebuilding config. Useful for "kick the container" when the process is stuck or the container needs a clean restart without changing the NixOS config. Rides the job queue (serialized against in-flight rebuilds for the same agent); waits with live progress unless `--no-wait` +* `restart-all` — Restart ALL managed agent containers via one restart DAG each — unrelated agents overlap, each serializes on its own lease. Waits for the whole set with live progress unless `--no-wait` @@ -290,21 +290,29 @@ Show all managed agents with their status (running / needs-login / needs-update) ## `hivectl agents restart` -Stop and start a single agent container without rebuilding config. Useful for "kick the container" when the process is stuck or the container needs a clean restart without changing the NixOS config +Stop and start a single agent container without rebuilding config. Useful for "kick the container" when the process is stuck or the container needs a clean restart without changing the NixOS config. Rides the job queue (serialized against in-flight rebuilds for the same agent); waits with live progress unless `--no-wait` -**Usage:** `hivectl agents restart ` +**Usage:** `hivectl agents restart [OPTIONS] ` ###### **Arguments:** * `` — Agent name (e.g. `damocles`, `ruth`) +###### **Options:** + +* `--no-wait` — Return immediately after the restart DAG is queued + ## `hivectl agents restart-all` -Stop and restart ALL managed agent containers in sequence. Iterates the live container list and restarts each one. Any per-agent failure is reported at the end rather than stopping mid-run, so all containers get a restart attempt +Restart ALL managed agent containers via one restart DAG each — unrelated agents overlap, each serializes on its own lease. Waits for the whole set with live progress unless `--no-wait` -**Usage:** `hivectl agents restart-all` +**Usage:** `hivectl agents restart-all [OPTIONS]` + +###### **Options:** + +* `--no-wait` — Return immediately after the restart DAGs are queued @@ -408,6 +416,7 @@ Stop containers hive-wide in one operator action. Bare `hivectl stop` stops **ev * `--gateway` — The gateway container (`hive-gateway`) * `--matrix` — The matrix container (`hive-matrix`) * `--graceful` — Gracefully quiesce each agent before stopping, instead of a hard stop. Each agent gets a graceful-stop DAG on the job queue: the harness is signalled, runs one stop-checkpoint turn to flush durable `/state`, drains, then the container is stopped (bounded by a 3-min timeout that falls back to a hard stop). All drains overlap. Applies to agents only +* `--no-wait` — Return immediately after the stop DAGs are queued instead of waiting for them with live per-node progress @@ -425,6 +434,7 @@ Start containers hive-wide — the inverse of `hivectl stop`. Bare `hivectl star * `--forge` — The forge container (`hive-forge`) * `--gateway` — The gateway container (`hive-gateway`) * `--matrix` — The matrix container (`hive-matrix`) +* `--no-wait` — Return immediately after the start DAGs are queued instead of waiting for them with live per-node progress diff --git a/hive-c0re/src/bin/hivectl.rs b/hive-c0re/src/bin/hivectl.rs index 9816f90b..515860a8 100644 --- a/hive-c0re/src/bin/hivectl.rs +++ b/hive-c0re/src/bin/hivectl.rs @@ -146,6 +146,10 @@ enum Cmd { /// hard stop). All drains overlap. Applies to agents only. #[arg(long)] graceful: bool, + /// Return immediately after the stop DAGs are queued instead of + /// waiting for them with live per-node progress. + #[arg(long)] + no_wait: bool, }, /// Start containers hive-wide — the inverse of `hivectl stop`. Bare /// `hivectl start` starts everything back up; the same scope flags as @@ -154,6 +158,10 @@ enum Cmd { Start { #[command(flatten)] scope: ScopeArgs, + /// Return immediately after the start DAGs are queued instead + /// of waiting for them with live per-node progress. + #[arg(long)] + no_wait: bool, }, /// Restart containers hive-wide — `stop` then `start` over the same /// scope. Bare `hivectl restart` restarts **everything** (all sub-agents @@ -531,15 +539,23 @@ enum AgentsCmd { /// Stop and start a single agent container without rebuilding config. /// Useful for "kick the container" when the process is stuck or the /// container needs a clean restart without changing the NixOS config. + /// Rides the job queue (serialized against in-flight rebuilds for + /// the same agent); waits with live progress unless `--no-wait`. Restart { /// Agent name (e.g. `damocles`, `ruth`). name: String, + /// Return immediately after the restart DAG is queued. + #[arg(long)] + no_wait: bool, + }, + /// Restart ALL managed agent containers via one restart DAG each — + /// unrelated agents overlap, each serializes on its own lease. + /// Waits for the whole set with live progress unless `--no-wait`. + RestartAll { + /// Return immediately after the restart DAGs are queued. + #[arg(long)] + no_wait: bool, }, - /// Stop and restart ALL managed agent containers in sequence. - /// Iterates the live container list and restarts each one. Any per-agent - /// failure is reported at the end rather than stopping mid-run, so all - /// containers get a restart attempt. - RestartAll, } #[derive(Subcommand)] @@ -602,8 +618,8 @@ async fn main() -> Result<()> { }, Cmd::Agents { cmd } => match cmd { AgentsCmd::List { json } => agents_list(&socket, json).await, - AgentsCmd::Restart { name } => agents_restart(&socket, &name).await, - AgentsCmd::RestartAll => agents_restart_all(&socket).await, + AgentsCmd::Restart { name, no_wait } => agents_restart(&socket, &name, no_wait).await, + AgentsCmd::RestartAll { no_wait } => agents_restart_all(&socket, no_wait).await, }, Cmd::Wg { cmd } => match cmd { WgCmd::Init { address } => wg_init(&socket, address.as_deref()).await, @@ -626,8 +642,12 @@ async fn main() -> Result<()> { peer_config(&domain, wg_address.as_deref(), wg_endpoint.as_deref()); Ok(()) } - Cmd::Stop { scope, graceful } => stop(&socket, scope.to_scope(), graceful).await, - Cmd::Start { scope } => start(&socket, scope.to_scope()).await, + Cmd::Stop { + scope, + graceful, + no_wait, + } => stop(&socket, scope.to_scope(), graceful, no_wait).await, + Cmd::Start { scope, no_wait } => start(&socket, scope.to_scope(), no_wait).await, Cmd::Restart { scope, graceful } => restart(&socket, scope.to_scope(), graceful).await, Cmd::Subvol { cmd } => match cmd { SubvolCmd::Upgrade { name, yes } => subvol_upgrade(&socket, &name, yes).await, @@ -1406,7 +1426,7 @@ fn gateway_list_users(file: &Path) -> Result<()> { // Agent management helpers (require daemon via host admin socket) // --------------------------------------------------------------------------- -async fn agents_restart(socket: &Path, name: &str) -> Result<()> { +async fn agents_restart(socket: &Path, name: &str, no_wait: bool) -> Result<()> { let resp = hive_c0re::client::request( socket, hive_sh4re::HostRequest::Restart { @@ -1416,8 +1436,8 @@ async fn agents_restart(socket: &Path, name: &str) -> Result<()> { .await .with_context(|| format!("connect to daemon socket {}", socket.display()))?; if resp.ok { - println!("restarted: {name}"); - Ok(()) + println!("restart queued: {name}"); + wait_for_dags(socket, resp.queued_dags.unwrap_or_default(), no_wait).await } else { bail!( "restart {name}: {}", @@ -1426,6 +1446,104 @@ async fn agents_restart(socket: &Path, name: &str) -> Result<()> { } } +// --------------------------------------------------------------------------- +// Job-queue wait/progress loop — shared by every verb that submits DAGs +// --------------------------------------------------------------------------- + +/// Poll the submitted DAG ids (`HostRequest::QueueDag`, ~1s interval) +/// and print a progress line whenever a DAG's rendered state changes — +/// including fan-out children that appear under a polled parent. Exits +/// non-zero when any DAG (or child) ends `failed`; a `cancelled` DAG +/// terminates the wait but is an operator action, not an error. +async fn wait_for_dags(socket: &Path, ids: Vec, no_wait: bool) -> Result<()> { + if no_wait || ids.is_empty() { + return Ok(()); + } + let mut pending: std::collections::BTreeSet = ids.into_iter().collect(); + let mut last: std::collections::HashMap = std::collections::HashMap::new(); + let mut failed: Vec = Vec::new(); + while !pending.is_empty() { + for id in pending.clone() { + let resp = hive_c0re::client::request(socket, hive_sh4re::HostRequest::QueueDag { id }) + .await + .with_context(|| format!("connect to daemon socket {}", socket.display()))?; + let dags = resp.dags.unwrap_or_default(); + if dags.is_empty() { + // Evicted from the queue's history tail — it finished a + // while ago; nothing left to report on. + println!("job #{id}: gone from queue history"); + pending.remove(&id); + continue; + } + let mut all_terminal = true; + for d in &dags { + let line = render_dag_line(d); + if last.get(&d.id) != Some(&line) { + println!("{line}"); + last.insert(d.id, line); + } + match d.state { + hive_sh4re::jobs::State::Failed => { + failed.push(format!("{} {}", d.kind.as_str(), d.agent)); + } + hive_sh4re::jobs::State::Done | hive_sh4re::jobs::State::Cancelled => {} + _ => all_terminal = false, + } + } + if all_terminal { + pending.remove(&id); + } + } + if !pending.is_empty() { + tokio::time::sleep(std::time::Duration::from_secs(1)).await; + } + } + if failed.is_empty() { + Ok(()) + } else { + failed.sort(); + failed.dedup(); + bail!("queued job(s) failed: {}", failed.join(", ")) + } +} + +fn state_glyph(state: hive_sh4re::jobs::State) -> &'static str { + match state { + hive_sh4re::jobs::State::Queued => "⏸", + hive_sh4re::jobs::State::Running => "▶", + hive_sh4re::jobs::State::Done => "✔", + hive_sh4re::jobs::State::Failed => "✖", + hive_sh4re::jobs::State::Cancelled => "⊘", + } +} + +/// One progress line for a DAG: roll-up glyph, template, agent, then +/// the node chain with the running node's live step label — the CLI +/// twin of the dashboard's queue card. +fn render_dag_line(d: &hive_sh4re::jobs::DagView) -> String { + use std::fmt::Write as _; + let mut out = format!( + "{} {} {:<12}", + state_glyph(d.state), + d.kind.as_str(), + d.agent + ); + for (i, n) in d.nodes.iter().enumerate() { + let sep = if i == 0 { " " } else { " → " }; + let _ = write!(out, "{sep}{} {}", state_glyph(n.state), n.kind); + if n.state == hive_sh4re::jobs::State::Running + && let Some(step) = &n.step + { + let _ = write!(out, " ({step})"); + } + } + if let Some(err) = d.nodes.iter().find_map(|n| n.error.as_deref()) { + let short: String = err.chars().take(120).collect(); + let _ = write!(out, " — {short}"); + } + out +} + /// `hivectl agents list` — fetch the per-agent status roster from the /// daemon (`HostRequest::AgentStatus`) and render it as a padded table, /// or the raw JSON rows with `--json`. Reuses the dashboard's @@ -1502,7 +1620,7 @@ async fn agents_list(socket: &Path, json: bool) -> Result<()> { Ok(()) } -async fn agents_restart_all(socket: &Path) -> Result<()> { +async fn agents_restart_all(socket: &Path, no_wait: bool) -> Result<()> { let resp = hive_c0re::client::request(socket, hive_sh4re::HostRequest::RestartAll) .await .with_context(|| format!("connect to daemon socket {}", socket.display()))?; @@ -1511,7 +1629,7 @@ async fn agents_restart_all(socket: &Path) -> Result<()> { println!("restart-all: no managed containers found"); } else { for a in agents { - println!("restarted: {a}"); + println!("restart queued: {a}"); } } if !resp.ok { @@ -1520,22 +1638,29 @@ async fn agents_restart_all(socket: &Path) -> Result<()> { resp.error.as_deref().unwrap_or("unknown error") ); } - Ok(()) + wait_for_dags(socket, resp.queued_dags.unwrap_or_default(), no_wait).await } -async fn stop(socket: &Path, scope: hive_sh4re::LifecycleScope, graceful: bool) -> Result<()> { +async fn stop( + socket: &Path, + scope: hive_sh4re::LifecycleScope, + graceful: bool, + no_wait: bool, +) -> Result<()> { let resp = hive_c0re::client::request(socket, hive_sh4re::HostRequest::Stop { scope, graceful }) .await .with_context(|| format!("connect to daemon socket {}", socket.display()))?; - render_lifecycle(&resp, "stopped") + render_lifecycle(&resp, "stop queued")?; + wait_for_dags(socket, resp.queued_dags.unwrap_or_default(), no_wait).await } -async fn start(socket: &Path, scope: hive_sh4re::LifecycleScope) -> Result<()> { +async fn start(socket: &Path, scope: hive_sh4re::LifecycleScope, no_wait: bool) -> Result<()> { let resp = hive_c0re::client::request(socket, hive_sh4re::HostRequest::Start { scope }) .await .with_context(|| format!("connect to daemon socket {}", socket.display()))?; - render_lifecycle(&resp, "started") + render_lifecycle(&resp, "start queued")?; + wait_for_dags(socket, resp.queued_dags.unwrap_or_default(), no_wait).await } /// Restart = `stop` then `start` over the same scope, composed client-side @@ -1543,9 +1668,13 @@ async fn start(socket: &Path, scope: hive_sh4re::LifecycleScope) -> Result<()> { /// `--graceful`; if it reports a failure (`stop` returns `Err`) the `?` /// short-circuits before the start phase, so a half-stopped hive isn't /// blindly started over — the operator sees the stop errors and can recover. +/// +/// No `--no-wait` here on purpose: the stop DAGs must complete before +/// the start submits, otherwise the start's `wanted = Up` write would +/// land before the queued stops execute and turn them into noops. async fn restart(socket: &Path, scope: hive_sh4re::LifecycleScope, graceful: bool) -> Result<()> { - stop(socket, scope.clone(), graceful).await?; - start(socket, scope).await + stop(socket, scope.clone(), graceful, false).await?; + start(socket, scope, false).await } /// A [`LifecycleScope`](hive_sh4re::LifecycleScope) targeting exactly one @@ -1673,3 +1802,82 @@ fn validate_htpasswd_username(username: &str) -> Result<()> { } Ok(()) } + +#[cfg(test)] +mod tests { + use hive_sh4re::jobs::{DagView, NodeView, Source, State, Template}; + + use super::render_dag_line; + + fn node(id: u32, kind: &str, state: State, step: Option<&str>) -> NodeView { + NodeView { + id, + kind: kind.to_owned(), + deps: if id == 0 { vec![] } else { vec![id - 1] }, + state, + step: step.map(str::to_owned), + build_log_id: None, + started_at: None, + finished_at: None, + error: None, + } + } + + #[test] + fn render_dag_line_shows_chain_and_running_step() { + let dag = DagView { + id: 7, + agent: "alice".to_owned(), + kind: Template::Rebuild, + state: State::Running, + source: Source::Manual, + parent_id: None, + reason: "manual".to_owned(), + enqueued_at: 0, + started_at: Some(1), + finished_at: None, + inputs: vec![], + approval_id: None, + perm_payload: None, + nodes: vec![ + node(0, "prebuild", State::Done, None), + node(1, "stop_for_update", State::Done, None), + node(2, "swap", State::Running, Some("nixos-container update")), + node(3, "reconcile", State::Queued, None), + ], + }; + let line = render_dag_line(&dag); + assert!(line.starts_with("▶ rebuild alice"), "{line}"); + assert!( + line.contains( + "✔ prebuild → ✔ stop_for_update → ▶ swap (nixos-container update) → ⏸ reconcile" + ), + "{line}" + ); + } + + #[test] + fn render_dag_line_surfaces_first_node_error() { + let mut failed = node(0, "prebuild", State::Failed, None); + failed.error = Some("nix build exploded".to_owned()); + let dag = DagView { + id: 8, + agent: "bob".to_owned(), + kind: Template::Rebuild, + state: State::Failed, + source: Source::Manual, + parent_id: None, + reason: "manual".to_owned(), + enqueued_at: 0, + started_at: Some(1), + finished_at: Some(2), + inputs: vec![], + approval_id: None, + perm_payload: None, + nodes: vec![failed], + }; + let line = render_dag_line(&dag); + assert!(line.contains("✖ rebuild"), "{line}"); + assert!(line.contains("— nix build exploded"), "{line}"); + } +} diff --git a/hive-c0re/src/job_queue/mod.rs b/hive-c0re/src/job_queue/mod.rs index b38dca51..f15ebe2c 100644 --- a/hive-c0re/src/job_queue/mod.rs +++ b/hive-c0re/src/job_queue/mod.rs @@ -169,7 +169,10 @@ impl JobQueue { && d.parent_id == spec.parent_id && d.approval_id == spec.approval_id && (d.template != Template::MetaUpdate || d.inputs == spec.inputs) - && PermPayload::same_type(d.perm_payload.as_ref(), spec.perm_payload.as_ref()) + && model::perm_payload_same_type( + d.perm_payload.as_ref(), + spec.perm_payload.as_ref(), + ) }) } diff --git a/hive-c0re/src/job_queue/model.rs b/hive-c0re/src/job_queue/model.rs index 39933504..c572e7ae 100644 --- a/hive-c0re/src/job_queue/model.rs +++ b/hive-c0re/src/job_queue/model.rs @@ -1,174 +1,37 @@ -//! Data model for the generic job-DAG queue: templates (what a DAG -//! *means*), node kinds (the primitive operations), dependency edges, -//! states, and the wire-facing `DagView` / `NodeView` snapshot shapes. +//! Data model for the generic job-DAG queue: node kinds (the primitive +//! operations), dependency edges, and the runtime `Dag` / `Node` store. +//! The serialized *views* — `DagView` / `NodeView` plus the `Template` +//! / `Source` / `State` / `PermPayload` wire enums — live in +//! `hive_sh4re::jobs` (wire types belong to the shared crate) and are +//! re-exported here for the queue's internal use. //! //! Two levels: the **DAG** is the unit of dedup / cancel / //! approval-resolution and the dashboard group; the **node** is the //! unit of scheduling / execution / build-log / step label. See //! `docs/coordinator.md::Job queue` for the full design. -use serde::{Deserialize, Serialize}; +pub use hive_sh4re::jobs::{DagView, NodeId, NodeView, PermPayload, Source, State, Template}; +use serde::Serialize; -/// What a DAG *means* — the request-level shape. Wire strings match the -/// old `QueueKind` values (serialized as the `kind` field on `DagView`) -/// so the dashboard's glyph map keeps working. -#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize)] -#[serde(rename_all = "snake_case")] -pub enum Template { - /// Rebuild one agent's container: `Prebuild → StopForUpdate → Swap - /// → Reconcile` (the tail `Reconcile` runs after `Swap` terminal, - /// ok *or* fail — the recovery-start). - Rebuild, - /// Bump meta flake locks: one `MetaLock` node; child `Rebuild` - /// DAGs fan out on completion for every affected agent. - MetaUpdate, - /// First-deploy spawn (approval-driven): `Create → WriteDropin → - /// Reconcile`. - Spawn, - /// Reserved for a future destroy integration — kept so the wire - /// shape doesn't need to change later. - #[allow(dead_code, reason = "wire shape — routed by a future PR")] - Destroy, - /// Boot-time config sweep: one `MetaLock` (hyperhive input, - /// non-fatal) node; stale agents' `Rebuild` DAGs fan out on - /// completion. - StartupSweep, - /// `StopForUpdate → Reconcile` with `wanted` set to `Up` at submit - /// time — a mechanical stop + start, like the old - /// `lifecycle::restart`, regardless of prior intent drift. - Restart, - /// `WritePermFile → Prebuild → StopForUpdate → Swap → Reconcile` — - /// perm-file commit followed by the rebuild subgraph. - PermChange, - /// `Signal → Drain → Reconcile` with `wanted` set to `Offline` at - /// submit time: quiesce the harness, await the drain (bounded), - /// then the tail `Reconcile` performs the actual container stop. - GracefulStop, - /// Single `Reconcile` with `wanted` set to `Up` at submit time. - Start, - /// Single `Reconcile` with `wanted` set to `Offline` at submit time. - Stop, - /// Single `Reconcile` with `wanted` untouched — boot-time converge - /// of observed state to the persisted intent. - Reconcile, +/// Dedup compares the perm *type*, not the value — a tool-groups +/// change and a capabilities change for the same agent are distinct +/// operations that must not collapse. +pub(super) fn perm_payload_same_type(a: Option<&PermPayload>, b: Option<&PermPayload>) -> bool { + matches!( + (a, b), + ( + Some(PermPayload::ToolGroups { .. }), + Some(PermPayload::ToolGroups { .. }) + ) | ( + Some(PermPayload::Capabilities { .. }), + Some(PermPayload::Capabilities { .. }) + ) | ( + Some(PermPayload::Combined { .. }), + Some(PermPayload::Combined { .. }) + ) | (None, None) + ) } -impl Template { - pub fn as_str(self) -> &'static str { - match self { - Template::Rebuild => "rebuild", - Template::MetaUpdate => "meta_update", - Template::Spawn => "spawn", - Template::Destroy => "destroy", - Template::StartupSweep => "startup_sweep", - Template::Restart => "restart", - Template::PermChange => "perm_change", - Template::GracefulStop => "graceful_stop", - Template::Start => "start", - Template::Stop => "stop", - Template::Reconcile => "reconcile", - } - } -} - -/// Where the submit request originated. Same variants + wire strings -/// as the old `QueueSource` — drives the "why" chip on the dashboard. -#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)] -#[serde(rename_all = "snake_case")] -pub enum Source { - /// Operator action (dashboard button, CLI, manager tool). - Manual, - /// Cascade child of a `MetaUpdate` DAG's fan-out; `parent_id` - /// points back at the originating meta-update. - MetaUpdate, - /// Boot-time submission (sweep parent, boot reconciles). - AutoUpdate, - /// Cascade child of a `StartupSweep` DAG's fan-out. - StartupSweep, - /// Crash recovery path (future use). - #[allow(dead_code, reason = "wire shape — used by a future feature")] - CrashRecover, - /// Operator approved a pending `Approval` row; `approval_id` on - /// the DAG points back at the source row. - Approval, -} - -impl Source { - pub fn as_str(self) -> &'static str { - match self { - Source::Manual => "manual", - Source::MetaUpdate => "meta_update", - Source::AutoUpdate => "auto_update", - Source::StartupSweep => "startup_sweep", - Source::CrashRecover => "crash_recover", - Source::Approval => "approval", - } - } -} - -/// Lifecycle state of a node — and, rolled up, of a DAG. Same wire -/// strings as the old `QueueState`. -#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)] -#[serde(rename_all = "snake_case")] -pub enum State { - Queued, - Running, - Done, - Failed, - Cancelled, -} - -impl State { - pub fn is_terminal(self) -> bool { - matches!(self, State::Done | State::Failed | State::Cancelled) - } -} - -/// Kind-specific payload for `Template::PermChange` DAGs. Carried on -/// the DAG (not the node) so dedup can compare the perm *type* -/// discriminant. Identical to the old `rebuild_queue::PermPayload`. -#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] -#[serde(tag = "type", rename_all = "snake_case")] -pub enum PermPayload { - /// Set the tool groups for one agent (`tool-groups.json`). - ToolGroups { groups: Vec }, - /// Set the capabilities for one agent (`capabilities.json`). - Capabilities { caps: Vec }, - /// Set both perm-types in one entry — the batch - /// `POST /api/permissions` path. `None` leaves that file untouched; - /// the worker commits whichever are present in one git commit, - /// then rebuilds once. - Combined { - groups: Option>, - caps: Option>, - }, -} - -impl PermPayload { - /// Dedup compares the perm *type*, not the value — a tool-groups - /// change and a capabilities change for the same agent are - /// distinct operations that must not collapse. - pub fn same_type(a: Option<&PermPayload>, b: Option<&PermPayload>) -> bool { - matches!( - (a, b), - ( - Some(PermPayload::ToolGroups { .. }), - Some(PermPayload::ToolGroups { .. }) - ) | ( - Some(PermPayload::Capabilities { .. }), - Some(PermPayload::Capabilities { .. }) - ) | ( - Some(PermPayload::Combined { .. }), - Some(PermPayload::Combined { .. }) - ) | (None, None) - ) - } -} - -/// Node id, unique within its DAG (dense small ints assigned by the -/// template builders). -pub type NodeId = u32; - /// When a dependency edge is considered satisfied. #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)] #[serde(rename_all = "snake_case")] @@ -427,58 +290,6 @@ impl Dag { } } -/// Wire shape of one node inside a `DagView`. -#[derive(Debug, Clone, Serialize)] -pub struct NodeView { - pub id: NodeId, - /// Flattened `NodeKind` tag ("prebuild", "swap", …). - pub kind: &'static str, - /// Ids of the nodes this one waits for. - pub deps: Vec, - pub state: State, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub step: Option, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub build_log_id: Option, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub started_at: Option, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub finished_at: Option, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub error: Option, -} - -/// Wire shape of a DAG, serialized onto `RebuildQueueChanged` and the -/// `/api/state` snapshot. DAG-level fields mirror the old `QueueEntry` -/// names (`kind` = template string, roll-up `state`); everything -/// per-node — step labels, build-log links, errors, timestamps — -/// appears exactly once, inside `nodes`. -#[derive(Debug, Clone, Serialize)] -pub struct DagView { - pub id: u64, - pub agent: String, - /// Template wire string — same values the old `kind` field used. - pub kind: Template, - /// Roll-up state (see [`Dag::rollup`]). - pub state: State, - pub source: Source, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub parent_id: Option, - pub reason: String, - pub enqueued_at: i64, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub started_at: Option, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub finished_at: Option, - #[serde(default, skip_serializing_if = "Vec::is_empty")] - pub inputs: Vec, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub approval_id: Option, - #[serde(default, skip_serializing_if = "Option::is_none")] - pub perm_payload: Option, - pub nodes: Vec, -} - impl Dag { pub fn view(&self) -> DagView { let started_at = self.nodes.iter().filter_map(|n| n.started_at).min(); @@ -506,7 +317,7 @@ impl Dag { .iter() .map(|n| NodeView { id: n.id, - kind: n.kind.as_str(), + kind: n.kind.as_str().to_owned(), deps: n.deps.iter().map(|d| d.on).collect(), state: n.state, step: n.step.clone(), diff --git a/hive-c0re/src/lifecycle/mod.rs b/hive-c0re/src/lifecycle/mod.rs index 7097aed4..b2c90733 100644 --- a/hive-c0re/src/lifecycle/mod.rs +++ b/hive-c0re/src/lifecycle/mod.rs @@ -501,59 +501,14 @@ pub async fn destroy(name: &str) -> Result<()> { Ok(()) } -/// Rebuild `name`'s container: sync the meta flake, optionally re-lock -/// the agent's input, then re-apply + restart via `nixos-container`. -/// -/// When `relock` is `true` the agent's meta input is bumped to whatever -/// `applied//main` points at before the build. Pass `false` for -/// meta-update cascade rebuilds, where re-locking would revert the bump -/// the cascade just committed (see the inline note below). -/// -/// # Errors -/// -/// Propagates errors from meta-flake sync / lock-update and the -/// `nixos-container` apply + restart shellouts. -/// -/// Returns `true` when `defer_start` suppressed the start-after-update — -/// the caller owns bringing the container back up (see -/// [`rebuild_no_meta`]). -pub async fn rebuild( - name: &str, - hive: &HiveEnv, - paths: &AgentPaths, - relock: bool, - defer_start: bool, - on_step: &(dyn Fn(&str) + Send + Sync), - on_build_log_id: &(dyn Fn(i64) + Send + Sync), -) -> Result { - // Sync the meta flake (idempotent — no-op when the rendered - // flake matches disk) so a manual rebuild from the dashboard - // can also recover from a divergent meta repo (e.g. an agent - // got added directly via `nixos-container create` outside - // hive-c0re). - let agents = agents_for_meta(None).await?; - crate::meta::sync_agents(hive, &agents).await?; - // Then bump just this agent's input — picks up whatever - // `applied//main` currently points at (deployed/). - // Commits the lock if it changed. - // - // `relock = false` skips this: a meta-update cascade has *just* set - // the meta lock deliberately, and `lock_update_for_rebuild` re-runs - // `nix flake update agent-`, which re-resolves the agent's - // transitive inputs back to the agent's own flake.lock — reverting - // the input the meta-update just bumped. Cascade rebuilds therefore - // build against the freshly-set on-disk lock as-is. - if relock { - crate::meta::lock_update_for_rebuild(name).await?; - } - rebuild_no_meta(name, hive, paths, defer_start, on_step, on_build_log_id).await -} - -/// Container-level rebuild without touching the meta repo. Callers -/// that own the meta side themselves (`actions::run_apply_commit` -/// drives meta through the two-phase prepare/finalize/abort flow) -/// use this directly. Public `rebuild` wraps it with idempotent meta -/// sync + lock-bump-and-commit. +/// Container-level rebuild without touching the meta repo. The one +/// remaining fused stop/update/start pipeline: the approval deploy +/// (`actions::deploy_applied_target`) drives meta through the +/// two-phase prepare/finalize/abort flow itself and needs the inline +/// start to verify the agent comes back up before finalizing. Every +/// other rebuild is a job-queue DAG (`Prebuild → StopForUpdate → Swap +/// → Reconcile`) whose `Prebuild` executor owns the meta sync + +/// relock this path's deleted `rebuild` wrapper used to do. /// /// `on_step` is called at each phase boundary with a short human-readable /// label so callers can surface progress (e.g. update the rebuild-queue diff --git a/hive-c0re/src/server.rs b/hive-c0re/src/server.rs index c64be3e1..b60705aa 100644 --- a/hive-c0re/src/server.rs +++ b/hive-c0re/src/server.rs @@ -90,21 +90,8 @@ async fn dispatch(req: &HostRequest, coord: Arc) -> HostResponse { tracing::info!(%id, %name, "spawn approval queued"); HostResponse::success() } - HostRequest::Kill { name } => handle_kill(&coord, name).await?, - HostRequest::Restart { name } => { - tracing::info!(%name, "restart"); - // Through the queue: serializes against in-flight - // rebuilds via the agent lease, writes `wanted = Up`, - // and gets the transient/crash-watch suppression the - // direct kill+start lacked. Returns once queued. - crate::job_queue::submit::restart( - &coord, - name, - crate::job_queue::Source::Manual, - "manual restart via hivectl".to_owned(), - ); - HostResponse::success() - } + HostRequest::Kill { name } => submit_single(&coord, name, Verb::Kill), + HostRequest::Restart { name } => submit_single(&coord, name, Verb::Restart), HostRequest::RestartAll => handle_restart_all(&coord).await?, HostRequest::Stop { scope, graceful } => { // Resolve the scope to explicit container names at the entry @@ -146,7 +133,17 @@ async fn dispatch(req: &HostRequest, coord: Arc) -> HostResponse { actions::destroy(&coord, name, *purge).await?; HostResponse::success() } - HostRequest::Rebuild { name } => handle_rebuild(&coord, name).await?, + HostRequest::Rebuild { name } => submit_single(&coord, name, Verb::Rebuild), + HostRequest::QueueDag { id } => { + // The polled DAG first, then its live fan-out children. + let dags = coord + .job_queue + .snapshot() + .into_iter() + .filter(|d| d.id == *id || d.parent_id == Some(*id)) + .collect(); + HostResponse::dags(dags) + } HostRequest::List => HostResponse::list(lifecycle::list().await?), HostRequest::AgentStatus => { let rows = crate::container_view::build_all(&coord) @@ -235,19 +232,55 @@ async fn handle_spawn(coord: &Arc, name: &str) -> Result, name: &str) -> Result { - tracing::info!(%name, "kill"); - if let Err(e) = coord.power.set(name, crate::power::Wanted::Offline) { - tracing::warn!(%name, error = ?e, "agent_power: set wanted=offline failed"); - } - lifecycle::kill(name).await?; - coord.unregister_agent(name); - coord.notify_manager(&hive_sh4re::HelperEvent::Killed { - agent: name.to_owned(), - }); - Ok(HostResponse::success()) +/// Single-agent queue verbs the admin socket exposes. Each submits the +/// matching DAG (persisting the `wanted` intent, serializing on the +/// agent's lease, with the transient/crash-watch suppression the old +/// direct lifecycle calls lacked) and returns the DAG id for the +/// client's wait loop. +#[derive(Clone, Copy)] +enum Verb { + /// Stop DAG (`wanted = Offline`; Reconcile kills + unregisters + + /// fires `Killed`). + Kill, + /// Restart DAG (`wanted = Up`; mechanical stop + reconcile-start). + Restart, + /// Rebuild DAG — the Swap tail owns the manager `Rebuilt` events + + /// kick, so the CLI path can't drift from the dashboard's. + Rebuild, +} + +fn submit_single(coord: &Arc, name: &str, verb: Verb) -> HostResponse { + use crate::job_queue::{Source, submit}; + let id = match verb { + Verb::Kill => { + tracing::info!(%name, "kill"); + submit::stop( + coord, + name, + Source::Manual, + "manual kill via hivectl".to_owned(), + ) + } + Verb::Restart => { + tracing::info!(%name, "restart"); + submit::restart( + coord, + name, + Source::Manual, + "manual restart via hivectl".to_owned(), + ) + } + Verb::Rebuild => { + tracing::info!(%name, "rebuild"); + submit::rebuild( + coord, + name, + Source::Manual, + "manual rebuild via hivectl".to_owned(), + ) + } + }; + HostResponse::queued(vec![id]) } /// Restart every container by submitting one restart DAG per agent — @@ -280,11 +313,13 @@ async fn handle_restart_all(coord: &Arc) -> Result { /// `handle_restart_all`. Callers resolve the [`LifecycleScope`] to these /// explicit name lists up front — this never sees the "all" flag. /// -/// A `graceful` stop enqueues a `QueueKind::GracefulStop` per agent (signal the -/// harness, run one stop-checkpoint turn, drain, then container stop, with a -/// timeout fallback to a hard stop), mirroring the dashboard `?graceful=1` -/// path. `graceful` applies to agents only - infra containers have no harness -/// turn loop, so they're always hard-stopped. +/// Every agent rides the job queue: a `graceful` stop submits the +/// quiesce DAG (signal → drain → reconcile-stop; all drains overlap), +/// a hard stop a plain stop DAG — both persist `wanted = Offline` and +/// serialize on the agent's lease so nothing races an in-flight +/// rebuild. The response carries the DAG ids so `hivectl` can wait +/// with per-node progress. Infra containers have no harness / lease +/// and stay direct + synchronous. async fn handle_stop( coord: &Arc, agents: &[String], @@ -294,36 +329,31 @@ async fn handle_stop( tracing::info!(?agents, ?infra, graceful, "stop"); let mut ok_items: Vec = Vec::new(); let mut errors: Vec = Vec::new(); + let mut queued: Vec = Vec::new(); for agent in agents { - if graceful { - // Graceful stop: submit the quiesce DAG rather than a hard - // kill. The per-agent lease keeps it from racing an - // in-flight rebuild for the same agent; the cheap Signal - // nodes all fire immediately so every agent's drain - // overlaps. `submit::graceful_stop` also persists - // `wanted = Offline` and emits the queue snapshot. + let reason = if graceful { + "manual via hivectl graceful stop" + } else { + "manual via hivectl stop" + }; + let id = if graceful { crate::job_queue::submit::graceful_stop( coord, agent, crate::job_queue::Source::Manual, - "manual via hivectl graceful stop".to_owned(), - ); - ok_items.push(agent.clone()); - continue; - } - // Persist the intent even if the kill itself fails — otherwise - // the next boot reconcile would restart the agent. - if let Err(e) = coord.power.set(agent, crate::power::Wanted::Offline) { - tracing::warn!(%agent, error = ?e, "agent_power: set wanted=offline failed"); - } - match lifecycle::kill(agent).await { - Ok(()) => ok_items.push(agent.clone()), - Err(e) => { - tracing::warn!(%agent, error = ?e, "stop: agent kill failed"); - errors.push(format!("{agent}: {e:#}")); - } - } + reason.to_owned(), + ) + } else { + crate::job_queue::submit::stop( + coord, + agent, + crate::job_queue::Source::Manual, + reason.to_owned(), + ) + }; + queued.push(id); + ok_items.push(agent.clone()); } for &container in infra { @@ -337,7 +367,9 @@ async fn handle_stop( } } - Ok(finish_lifecycle(ok_items, &errors)) + let mut resp = finish_lifecycle(ok_items, &errors); + resp.queued_dags = Some(queued); + Ok(resp) } /// Start the given `infra` containers then `agents` (`hivectl start`) — the @@ -364,22 +396,23 @@ async fn handle_start( } } + let mut queued: Vec = Vec::new(); for agent in agents { - // Persist the intent even if the start itself fails — the next - // reconcile (boot or queued) retries toward `Up`. - if let Err(e) = coord.power.set(agent, crate::power::Wanted::Up) { - tracing::warn!(%agent, error = ?e, "agent_power: set wanted=up failed"); - } - match lifecycle::start(agent).await { - Ok(()) => ok_items.push(agent.clone()), - Err(e) => { - tracing::warn!(%agent, error = ?e, "start: agent start failed"); - errors.push(format!("{agent}: {e:#}")); - } - } + // Through the queue: persists `wanted = Up`, upgrades a + // stale-rev start to a full rebuild, and serializes on the + // agent's lease. Ids ride back for hivectl's wait loop. + queued.push(crate::job_queue::submit::start( + coord, + agent, + crate::job_queue::Source::Manual, + "manual via hivectl start".to_owned(), + )); + ok_items.push(agent.clone()); } - Ok(finish_lifecycle(ok_items, &errors)) + let mut resp = finish_lifecycle(ok_items, &errors); + resp.queued_dags = Some(queued); + Ok(resp) } /// Resolve which sub-agent logical names a scope targets: every live @@ -464,48 +497,7 @@ fn finish_lifecycle(ok_items: Vec, errors: &[String]) -> HostResponse { ok: false, error: Some(errors.join("; ")), agents: Some(ok_items), - approvals: None, - urls: None, - agent_statuses: None, + ..HostResponse::default() } } } - -/// Rebuild `name`'s container, notifying the manager of the outcome -/// (success or failure) and kicking the agent's next turn on success. -async fn handle_rebuild(coord: &Arc, name: &str) -> Result { - tracing::info!(%name, "rebuild"); - let agent_dir = coord.ensure_runtime(name)?; - let hive = coord.hive_env(); - let paths = Coordinator::agent_paths(name, agent_dir); - let result = lifecycle::rebuild(name, &hive, &paths, true, false, &|_| (), &|_| ()).await; - // Mirror auto_update::rebuild_agent — the manager wants to know - // about every rebuild attempt regardless of which surface triggered - // it, especially failures (build error → manager can adjust the - // agent's agent.nix). Without this the admin-socket CLI was a - // notify-gap. - match &result { - Ok(_) => { - coord.notify_manager(&hive_sh4re::HelperEvent::Rebuilt { - agent: name.to_owned(), - ok: true, - note: None, - sha: None, - tag: None, - }); - // Wake the agent's next turn with the "you were rebuilt" - // hint. Same pattern as auto_update::rebuild_agent and the - // dashboard rebuild path — this is the CLI's equivalent. - coord.kick_agent(name, "container rebuilt"); - } - Err(e) => coord.notify_manager(&hive_sh4re::HelperEvent::Rebuilt { - agent: name.to_owned(), - ok: false, - note: Some(format!("{e:#}")), - sha: None, - tag: None, - }), - } - result?; - Ok(HostResponse::success()) -} diff --git a/hive-sh4re/src/jobs.rs b/hive-sh4re/src/jobs.rs new file mode 100644 index 00000000..fe4a8fdd --- /dev/null +++ b/hive-sh4re/src/jobs.rs @@ -0,0 +1,191 @@ +//! Wire shapes of hive-c0re's job-DAG queue: what a queued job looks +//! like on the dashboard SSE channel (`rebuild_queue_changed`), the +//! `/api/state.rebuild_queue` snapshot, and the host admin socket's +//! `QueueDag` polling surface (`hivectl`'s wait/progress loop). The +//! queue *internals* — node kinds, dependency edges, scheduling state — +//! live in `hive-c0re::job_queue`; these are the serialized views it +//! produces. Semantics: `docs/coordinator.md::Job queue`. + +use serde::{Deserialize, Serialize}; + +/// What a DAG *means* — the request-level shape. Wire strings match +/// the pre-DAG queue's `kind` values so dashboards key off the same +/// tags. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum Template { + /// Rebuild one agent's container (prebuild → stop → profile-swap → + /// reconcile). + Rebuild, + /// Bump meta flake locks; child `Rebuild` DAGs fan out on + /// completion for every affected agent. + MetaUpdate, + /// First-deploy spawn (approval-driven). + Spawn, + /// Reserved for a future destroy integration. + Destroy, + /// Boot-time config sweep (hyperhive lock bump + stale-agent + /// rebuild fan-out). + StartupSweep, + /// Mechanical stop + converge to `wanted = Up` (a restart). + Restart, + /// Perm-file commit followed by the rebuild subgraph. + PermChange, + /// Quiesce the harness, drain, then stop (`wanted = Offline`). + GracefulStop, + /// Converge to `wanted = Up`. + Start, + /// Converge to `wanted = Offline`. + Stop, + /// Bare converge of observed power state to the persisted intent + /// (boot reconcile). + Reconcile, +} + +impl Template { + #[must_use] + pub fn as_str(self) -> &'static str { + match self { + Template::Rebuild => "rebuild", + Template::MetaUpdate => "meta_update", + Template::Spawn => "spawn", + Template::Destroy => "destroy", + Template::StartupSweep => "startup_sweep", + Template::Restart => "restart", + Template::PermChange => "perm_change", + Template::GracefulStop => "graceful_stop", + Template::Start => "start", + Template::Stop => "stop", + Template::Reconcile => "reconcile", + } + } +} + +/// Where the submit request originated — drives the "why" chip on the +/// dashboard. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum Source { + /// Operator action (dashboard button, CLI, manager tool). + Manual, + /// Cascade child of a `MetaUpdate` DAG's fan-out. + MetaUpdate, + /// Boot-time submission (sweep parent, boot reconciles). + AutoUpdate, + /// Cascade child of a `StartupSweep` DAG's fan-out. + StartupSweep, + /// Crash recovery path (future use). + CrashRecover, + /// Operator approved a pending `Approval` row; `approval_id` on + /// the DAG points back at the source row. + Approval, +} + +impl Source { + #[must_use] + pub fn as_str(self) -> &'static str { + match self { + Source::Manual => "manual", + Source::MetaUpdate => "meta_update", + Source::AutoUpdate => "auto_update", + Source::StartupSweep => "startup_sweep", + Source::CrashRecover => "crash_recover", + Source::Approval => "approval", + } + } +} + +/// Lifecycle state of a node — and, rolled up, of a DAG. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum State { + Queued, + Running, + Done, + Failed, + Cancelled, +} + +impl State { + #[must_use] + pub fn is_terminal(self) -> bool { + matches!(self, State::Done | State::Failed | State::Cancelled) + } +} + +/// Kind-specific payload for `Template::PermChange` DAGs. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(tag = "type", rename_all = "snake_case")] +pub enum PermPayload { + /// Set the tool groups for one agent (`tool-groups.json`). + ToolGroups { groups: Vec }, + /// Set the capabilities for one agent (`capabilities.json`). + Capabilities { caps: Vec }, + /// Set both perm-types in one entry — the batch + /// `POST /api/permissions` path. `None` leaves that file untouched; + /// present fields commit together and rebuild once. + Combined { + groups: Option>, + caps: Option>, + }, +} + +/// Node id, unique within its DAG. +pub type NodeId = u32; + +/// One node of a queued DAG, as serialized. Step labels, build-log +/// links, errors, and timestamps are per-node; the DAG-level `state` +/// is a roll-up. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct NodeView { + pub id: NodeId, + /// Node primitive tag: `"prebuild"`, `"stop_for_update"`, + /// `"swap"`, `"create"`, `"meta_lock"`, `"reconcile"`, `"signal"`, + /// `"drain"`, `"write_dropin"`, `"write_perm_file"`, + /// `"approval_deploy"`. + pub kind: String, + /// Ids of the nodes this one waits for. + #[serde(default)] + pub deps: Vec, + pub state: State, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub step: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub build_log_id: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub started_at: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub finished_at: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub error: Option, +} + +/// A queued/running/recent DAG. DAG-level fields mirror the pre-DAG +/// `QueueEntry` names (`kind` = template string, roll-up `state`); +/// everything per-node appears exactly once, inside `nodes`. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct DagView { + pub id: u64, + pub agent: String, + /// Template wire string — same values the old `kind` field used. + pub kind: Template, + /// Roll-up: `failed` if any node failed, else `running` / + /// `queued` / `cancelled` / `done`. + pub state: State, + pub source: Source, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub parent_id: Option, + pub reason: String, + pub enqueued_at: i64, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub started_at: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub finished_at: Option, + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub inputs: Vec, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub approval_id: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub perm_payload: Option, + pub nodes: Vec, +} diff --git a/hive-sh4re/src/lib.rs b/hive-sh4re/src/lib.rs index 3561ee0f..0f40a40a 100644 --- a/hive-sh4re/src/lib.rs +++ b/hive-sh4re/src/lib.rs @@ -4,6 +4,7 @@ use chrono::{DateTime, Utc}; use serde::{Deserialize, Serialize}; pub mod assets; +pub mod jobs; pub mod paths; pub mod priv_proto; pub mod wire_time; @@ -69,6 +70,10 @@ pub enum HostRequest { /// matrix GUI disabled). Backs `hivectl open` + the federation /// peer-config block (which reads the bare `domain`). Urls, + /// Fetch one job-queue DAG (plus its live fan-out children, linked + /// via `parent_id`) by id — the polling surface behind `hivectl`'s + /// wait/progress loop. Result: [`HostResponse::dags`]. + QueueDag { id: u64 }, /// List pending approval requests. Pending, /// Approve a pending request by id; the action runs immediately. @@ -170,7 +175,7 @@ pub struct HiveUrls { pub matrix: Option, } -#[derive(Debug, Clone, Serialize, Deserialize)] +#[derive(Debug, Clone, Default, Serialize, Deserialize)] pub struct HostResponse { pub ok: bool, #[serde(default, skip_serializing_if = "Option::is_none")] @@ -188,6 +193,16 @@ pub struct HostResponse { /// request kind. #[serde(default, skip_serializing_if = "Option::is_none")] pub agent_statuses: Option>, + /// Ids of the job-queue DAGs this request submitted (rebuild / + /// restart / power ops). Clients poll them via + /// [`HostRequest::QueueDag`]; `None` for non-submitting requests. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub queued_dags: Option>, + /// `QueueDag` result — the requested DAG followed by its live + /// fan-out children ([`jobs::DagView`]). Empty when the DAG has + /// been evicted from the queue's history tail. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub dags: Option>, } /// One row in the approval queue. `commit_ref` is overloaded per @@ -298,11 +313,7 @@ impl HostResponse { pub fn success() -> Self { Self { ok: true, - error: None, - agents: None, - approvals: None, - urls: None, - agent_statuses: None, + ..Self::default() } } @@ -311,10 +322,7 @@ impl HostResponse { Self { ok: false, error: Some(message.into()), - agents: None, - approvals: None, - urls: None, - agent_statuses: None, + ..Self::default() } } @@ -322,11 +330,8 @@ impl HostResponse { pub fn list(agents: Vec) -> Self { Self { ok: true, - error: None, agents: Some(agents), - approvals: None, - urls: None, - agent_statuses: None, + ..Self::default() } } @@ -334,11 +339,8 @@ impl HostResponse { pub fn pending(approvals: Vec) -> Self { Self { ok: true, - error: None, - agents: None, approvals: Some(approvals), - urls: None, - agent_statuses: None, + ..Self::default() } } @@ -347,11 +349,8 @@ impl HostResponse { pub fn urls(urls: HiveUrls) -> Self { Self { ok: true, - error: None, - agents: None, - approvals: None, urls: Some(urls), - agent_statuses: None, + ..Self::default() } } @@ -360,11 +359,29 @@ impl HostResponse { pub fn agent_statuses(rows: Vec) -> Self { Self { ok: true, - error: None, - agents: None, - approvals: None, - urls: None, agent_statuses: Some(rows), + ..Self::default() + } + } + + /// A request that submitted job-queue DAGs — carries their ids for + /// the client's wait/progress loop. + #[must_use] + pub fn queued(ids: Vec) -> Self { + Self { + ok: true, + queued_dags: Some(ids), + ..Self::default() + } + } + + /// `QueueDag` result — the polled DAG + its live children. + #[must_use] + pub fn dags(dags: Vec) -> Self { + Self { + ok: true, + dags: Some(dags), + ..Self::default() } } }