From 860484a1935f1113665b6d50302fd211cd711021 Mon Sep 17 00:00:00 2001 From: atlas Date: Tue, 14 Jul 2026 23:34:56 +0200 Subject: [PATCH] refactor(#2439): build hive-wide stop/start/restart DAGs dynamically MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Hive-wide `stop` / `start` / `restart` emit ONE DAG with a per-agent subgraph each (concurrent on their own leases) instead of N DAGs — and each subgraph is now built dynamically from the agent's live running state rather than a fixed template shape: - online agent: the full stop→reconcile (restart: stop-for-update→reconcile) chain; `graceful` prepends signal→drain. - offline agent: just `SetWanted → Reconcile` (nothing to quiesce/stop; a restart of a down agent is really a start). The head `SetWanted` (intent) and tail `Reconcile` (convergence guarantee) are always present; only the mechanical `Signal`/`Drain`/`StopForUpdate` nodes are state-conditional. Keeping `Reconcile` in every shape closes the TOCTOU window — a race-up between the `is_running` read and node exec is still converged in-DAG (with `StopForUpdate`-noop as the backstop) — with no reliance on an external reconcile sweep. The state-aware assembly needs an async `is_running` read, so it moves out of the pure/sync `templates.rs` into `submit.rs`, layered as pure `*_chain(running)` → pure `*_spec(targets)` (the unit-test seam) → async `*_many` (reads live state + submits). `templates.rs` keeps only the shared pure primitives (`node`/`after_ok`/`rebuild_nodes`). Callers await the now-async submit fns (server, dashboard, socket_server). Tests exercise both the online and offline shapes via the pure `*_spec` seam. docs/coordinator.md shapes updated. --- docs/coordinator.md | 33 +- hive-c0re/src/dashboard/lifecycle_ops.rs | 12 +- hive-c0re/src/job_queue/submit.rs | 355 +++++++++++++++--- hive-c0re/src/job_queue/templates.rs | 173 ++------- hive-c0re/src/job_queue/tests.rs | 215 +++++++---- hive-c0re/src/server.rs | 107 +++--- .../src/socket_server/lifecycle_handlers.rs | 12 +- hive-c0re/src/socket_server/mod.rs | 2 +- 8 files changed, 574 insertions(+), 335 deletions(-) diff --git a/docs/coordinator.md b/docs/coordinator.md index 7b0d6cff..81b4e440 100644 --- a/docs/coordinator.md +++ b/docs/coordinator.md @@ -76,19 +76,32 @@ container build: The power ops write the durable `wanted` intent via a head `SetWanted` node (not a pre-submit side effect) — it holds the agent lease, so -intent-write + reconcile is atomic per-agent. `restart` takes an agent -*list*: a hive-wide `hivectl restart` / `restart-all` is ONE DAG with a -per-agent restart subgraph each (independent roots, run concurrently on -their own leases), not N separate DAGs. +intent-write + reconcile is atomic per-agent. The hive-wide power ops — +`restart`, `stop`, and `start` — take an agent *list*: a hive-wide `hivectl +restart` / `stop` / `start` is ONE DAG with a per-agent subgraph each +(independent roots, run concurrently on their own leases), not N separate +DAGs. + +**These are built dynamically from each agent's live running state** (an +async `lifecycle::is_running` read), so they live in `job_queue/submit.rs`, +not the pure/sync `templates.rs`. Per-agent shape rule: the head `SetWanted` +(intent) and the tail `Reconcile` (convergence guarantee — cheap, noops when +already converged) are ALWAYS present; only the *mechanical* nodes +(`Signal`/`Drain`/`StopForUpdate`) are state-conditional — skipped for a +*down* agent (nothing to quiesce/stop). Keeping `Reconcile` in every shape +closes the TOCTOU window: if an agent flips state between the `is_running` +read and node exec, the tail `Reconcile` still converges it in-DAG (with +`StopForUpdate`-noop as the backstop) — no reliance on an external reconcile +sweep. `start` folds the per-agent stale-rev upgrade in (a *down + stale* +agent's subgraph is a rebuild-then-start). ```text rebuild(a): Prebuild(a) → StopForUpdate(a) → Swap(a) →(after-any) Reconcile(a) -graceful-stop(a): SetWanted(a,Off) → Signal(a) → Drain(a) → Reconcile(a) -restart(a..): per agent: SetWanted(a,Up) → StopForUpdate(a) → Reconcile(a) (N subgraphs, 1 DAG) -graceful-restart(a): SetWanted(a,Up) → Signal(a) → Drain(a) → StopForUpdate(a) → Reconcile(a) -start(a): SetWanted(a,Up) → Reconcile(a) (stale rev ⇒ stale-start below) -stale-start(a): SetWanted(a,Up) → «rebuild subgraph» (prebuild noops — agent is down) -stop(a): SetWanted(a,Off) → Reconcile(a) +stop(a..): online a: SetWanted(a,Off) → [Signal→Drain→ if graceful] Reconcile(a) + offline a: SetWanted(a,Off) → Reconcile(a) (N subgraphs, 1 DAG) +restart(a..): online a: SetWanted(a,Up) → [Signal→Drain→ if graceful] StopForUpdate(a) → Reconcile(a) + offline a: SetWanted(a,Up) → Reconcile(a) (nothing to stop — it's a start) +start(a..): a: SetWanted(a,Up) → Reconcile(a) (down+stale ⇒ SetWanted(a,Up) → «rebuild subgraph») spawn(a): [wanted=Up at approve] Create(a) → WriteDropin(a) → Reconcile(a) perm-change(a): WritePermFile(a) → «rebuild subgraph» meta-update(inp): MetaLock(inp) → «fan-out rebuild(a) per affected agent» diff --git a/hive-c0re/src/dashboard/lifecycle_ops.rs b/hive-c0re/src/dashboard/lifecycle_ops.rs index c139feaa..8ae3c24c 100644 --- a/hive-c0re/src/dashboard/lifecycle_ops.rs +++ b/hive-c0re/src/dashboard/lifecycle_ops.rs @@ -65,7 +65,8 @@ pub(super) async fn post_kill( &logical, Source::Manual, "manual via dashboard graceful stop".to_owned(), - ); + ) + .await; return (StatusCode::OK, "ok").into_response(); } // Manager is stoppable from the dashboard like any other @@ -83,7 +84,8 @@ pub(super) async fn post_kill( &logical, Source::Manual, "manual via dashboard stop".to_owned(), - ); + ) + .await; (StatusCode::OK, "ok").into_response() } @@ -100,7 +102,8 @@ pub(super) async fn post_restart( &logical, Source::Manual, "manual via dashboard ↺ R3START button".to_owned(), - ); + ) + .await; (StatusCode::OK, "ok").into_response() } @@ -117,7 +120,8 @@ pub(super) async fn post_start( &logical, Source::Manual, "manual via dashboard start".to_owned(), - ); + ) + .await; (StatusCode::OK, "ok").into_response() } diff --git a/hive-c0re/src/job_queue/submit.rs b/hive-c0re/src/job_queue/submit.rs index 93b718ef..ed030013 100644 --- a/hive-c0re/src/job_queue/submit.rs +++ b/hive-c0re/src/job_queue/submit.rs @@ -1,15 +1,32 @@ //! Request-level submit API — the surface the dashboard POST handlers, -//! the MCP socket handlers, and `hivectl` paths call. The durable -//! `wanted` power intent is now written by a `SetWanted` DAG node at the -//! head of each power-op template (not a pre-submit side effect); the -//! only submit-time logic left is the stale-start *shape* decision -//! (`start` vs `stale_start`). Every helper emits a fresh queue snapshot -//! so the dashboard shows the new DAG immediately. +//! the MCP socket handlers, and `hivectl` paths call. +//! +//! The **power ops** (`stop` / `start` / `restart`) are built here, not in +//! `templates.rs`: each agent's subgraph shape depends on its *live* running +//! state, which needs an async `lifecycle::is_running` read that a pure/sync +//! template can't do. So these fns are async — they read each agent's state, +//! assemble a per-agent subgraph out of the shared pure primitives +//! (`templates::{node, after_ok, rebuild_nodes}`), and concatenate them into +//! ONE DAG (independent per-agent roots, concurrent on their own leases). +//! +//! Dynamic shape rule: the head `SetWanted(w)` (durable intent) and the tail +//! `Reconcile` (the convergence guarantee — cheap, noops when already +//! converged) are ALWAYS present; only the *mechanical* nodes +//! (`Signal`/`Drain`/`StopForUpdate`) are state-conditional (skipped for a +//! down agent — nothing to quiesce/stop). Keeping `Reconcile` in every shape +//! closes the TOCTOU window: if an agent flips state between the `is_running` +//! read and node execution, the tail `Reconcile` still converges it in-DAG, +//! with `StopForUpdate`-noop as the backstop — no reliance on an external +//! reconcile sweep. Every helper emits a fresh queue snapshot so the +//! dashboard shows the new DAG immediately. use std::sync::Arc; +use super::model::{DagSpec, Dep, NodeKind, NodeSpec, Template}; +use super::templates::{after_ok, node, rebuild_nodes}; use super::{Source, templates}; -use crate::coordinator::Coordinator; +use crate::coordinator::{Coordinator, TransientKind}; +use crate::lifecycle; fn submit_and_emit(coord: &Arc, spec: super::DagSpec) -> u64 { let id = coord @@ -27,73 +44,305 @@ pub fn rebuild(coord: &Arc, agent: &str, source: Source, reason: St submit_and_emit(coord, templates::rebuild(agent, source, reason, None, true)) } -/// Restart a single agent: mechanical stop + converge to `wanted = Up`. -/// The intent write is the template's head `SetWanted(Up)` node — it -/// matters when `wanted` drifted `Offline` under a running agent (an -/// operator asking for a restart plainly wants it running, not a stop). -/// Thin wrapper over [`restart_many`] with a one-agent slice. -pub fn restart(coord: &Arc, agent: &str, source: Source, reason: String) -> u64 { - restart_many(coord, &[agent.to_owned()], false, source, reason) +// ---- dynamic power-op DAG assembly ---------------------------------------- +// +// The pure per-agent chain builders below take `running` (and `stale`) +// explicitly so they stay pure + unit-testable without a live container; +// the async `*_many` fns read the real state via `lifecycle::is_running` +// then hand it in. Each chain uses LOCAL (0-based) deps; `concat_subgraphs` +// rebases them into one DAG. + +/// One agent's **stop** subgraph. `SetWanted(Off)` head + `Reconcile` tail +/// always; the graceful `Signal → Drain` quiesce only when the agent is +/// actually running (nothing to drain on a down container). The `Reconcile` +/// stays even for a down agent so a race-up between the state read and exec +/// is still stopped in-DAG. +fn stop_chain(agent: &str, graceful: bool, running: bool) -> Vec { + let mut n = vec![node(agent, NodeKind::SetWanted { up: false }, Vec::new())]; + if graceful && running { + n.push(node(agent, NodeKind::Signal, after_ok(0))); + n.push(node(agent, NodeKind::Drain, after_ok(1))); + n.push(node(agent, NodeKind::Reconcile, after_ok(2))); + } else { + n.push(node(agent, NodeKind::Reconcile, after_ok(0))); + } + n +} + +/// One agent's **start** subgraph. `SetWanted(Up)` head; a down + stale-rev +/// agent gets the rebuild subgraph (its tail `Reconcile` starts it on +/// current derivations), otherwise a plain `Reconcile` (which starts a down +/// agent and noops an already-running one). +fn start_chain(agent: &str, running: bool, stale: bool) -> Vec { + let mut n = vec![node(agent, NodeKind::SetWanted { up: true }, Vec::new())]; + if !running && stale { + // Rebuild subgraph rooted at the SetWanted head (base = 1, so + // `Prebuild` deps `after_ok(0)` = the head). + n.extend(rebuild_nodes(agent, true, 1)); + } else { + n.push(node(agent, NodeKind::Reconcile, after_ok(0))); + } + n +} + +/// One agent's **restart** subgraph. `SetWanted(Up)` head + `Reconcile` +/// tail; the stop portion (`Signal → Drain` when graceful, then +/// `StopForUpdate`) only when the agent is running — a restart of a down +/// agent is really a start (`SetWanted(Up) → Reconcile`). +fn restart_chain(agent: &str, graceful: bool, running: bool) -> Vec { + let mut n = vec![node(agent, NodeKind::SetWanted { up: true }, Vec::new())]; + if running { + let mut prev = 0u32; + if graceful { + n.push(node(agent, NodeKind::Signal, after_ok(prev))); + prev += 1; + n.push(node(agent, NodeKind::Drain, after_ok(prev))); + prev += 1; + } + n.push(node(agent, NodeKind::StopForUpdate, after_ok(prev))); + prev += 1; + n.push(node(agent, NodeKind::Reconcile, after_ok(prev))); + } else { + n.push(node(agent, NodeKind::Reconcile, after_ok(0))); + } + n +} + +/// Concatenate per-agent subgraphs (each with LOCAL 0-based deps) into one +/// node list, rebasing each subgraph's internal deps by its offset. A +/// subgraph root (empty deps — the `SetWanted` head) stays a root, so the +/// per-agent subgraphs are independent and run concurrently, each on its +/// own lease. +fn concat_subgraphs(chains: Vec>) -> Vec { + let mut out: Vec = Vec::new(); + for chain in chains { + let base = u32::try_from(out.len()).unwrap_or(u32::MAX); + for spec in chain { + let deps = spec + .deps + .into_iter() + .map(|d| Dep { + on: base + d.on, + when: d.when, + }) + .collect(); + out.push(NodeSpec { + agent: spec.agent, + kind: spec.kind, + deps, + }); + } + } + out +} + +/// Wrap assembled power-op `nodes` in a `DagSpec`. +fn power_dag( + template: Template, + transient: TransientKind, + source: Source, + reason: String, + nodes: Vec, +) -> DagSpec { + DagSpec { + template, + source, + reason, + parent_id: None, + approval_id: None, + inputs: Vec::new(), + perm_payload: None, + transient: Some(transient), + nodes, + } +} + +// The `*_spec` builders below are the PURE core the async `*_many` fns call +// after reading live state — they take the per-agent running (and stale) +// flags explicitly, so unit tests exercise the online/offline shapes without +// a live container. `*_many` = gather state + call `*_spec` + submit. + +/// Assemble the stop DAG from explicit `(agent, running)` targets. +pub(crate) fn stop_spec( + targets: &[(String, bool)], + graceful: bool, + source: Source, + reason: String, +) -> DagSpec { + let chains = targets + .iter() + .map(|(agent, running)| stop_chain(agent, graceful, *running)) + .collect(); + let template = if graceful { + Template::GracefulStop + } else { + Template::Stop + }; + power_dag( + template, + TransientKind::Stopping, + source, + reason, + concat_subgraphs(chains), + ) +} + +/// Assemble the start DAG from explicit `(agent, running, stale)` targets. +/// Transient is `Rebuilding` when any down+stale agent grew a rebuild +/// subgraph (crash-watch suppression during its Swap), else `Starting`; +/// applied per-agent at claim time, so each agent still shows its own pill. +pub(crate) fn start_spec( + targets: &[(String, bool, bool)], + source: Source, + reason: String, +) -> DagSpec { + let chains = targets + .iter() + .map(|(agent, running, stale)| start_chain(agent, *running, *stale)) + .collect(); + let any_rebuild = targets.iter().any(|(_, running, stale)| !running && *stale); + let transient = if any_rebuild { + TransientKind::Rebuilding + } else { + TransientKind::Starting + }; + power_dag( + Template::Start, + transient, + source, + reason, + concat_subgraphs(chains), + ) +} + +/// Assemble the restart DAG from explicit `(agent, running)` targets. +pub(crate) fn restart_spec( + targets: &[(String, bool)], + graceful: bool, + source: Source, + reason: String, +) -> DagSpec { + let chains = targets + .iter() + .map(|(agent, running)| restart_chain(agent, graceful, *running)) + .collect(); + let template = if graceful { + Template::GracefulRestart + } else { + Template::Restart + }; + power_dag( + template, + TransientKind::Restarting, + source, + reason, + concat_subgraphs(chains), + ) +} + +/// Restart a single agent. Thin wrapper over [`restart_many`]. +pub async fn restart(coord: &Arc, agent: &str, source: Source, reason: String) -> u64 { + restart_many(coord, &[agent.to_owned()], false, source, reason).await +} + +/// Graceful restart of a single agent (signal → drain → stop → reconcile, +/// when running). Thin wrapper over [`restart_many`] with `graceful = true`. +pub async fn graceful_restart( + coord: &Arc, + agent: &str, + source: Source, + reason: String, +) -> u64 { + restart_many(coord, &[agent.to_owned()], true, source, reason).await } /// Restart `agents` (one or many) in a **single** DAG — one per-agent -/// subgraph each, running concurrently on their own leases. `graceful` -/// prepends the signal→drain quiesce per agent. The whole hive-wide -/// `hivectl restart` / `restart-all` is now one DAG instead of N. -pub fn restart_many( +/// subgraph each, built dynamically from live running state and run +/// concurrently on their own leases. A running agent gets the stop→reconcile +/// chain (`graceful` prepends signal→drain); a down agent gets just +/// `SetWanted(Up) → Reconcile` (there's nothing to stop). The whole +/// hive-wide `hivectl restart` / `restart-all` is one DAG. +pub async fn restart_many( coord: &Arc, agents: &[String], graceful: bool, source: Source, reason: String, ) -> u64 { - submit_and_emit(coord, templates::restart(agents, graceful, source, reason)) -} - -/// Start: `SetWanted(Up)` (a DAG node now) then reconcile. A stale rev -/// marker upgrades the start to a rebuild-then-start (`stale_start`, whose -/// tail `Reconcile` does the start) so the container always comes up on -/// current derivations — the old fast-lane `run_start` upgrade, still a -/// submit-time *shape* decision (which template), while the intent write -/// itself is now the template's head node. -pub fn start(coord: &Arc, agent: &str, source: Source, reason: String) -> u64 { - let stored = std::fs::read_to_string(crate::paths::applied_rev_marker(agent)).ok(); - let stale = crate::auto_update::current_flake_rev(&coord.hyperhive_flake) - .is_some_and(|rev| stored.as_deref() != Some(rev.as_str())); - if stale { - tracing::info!(%agent, "start: rev stale — upgrading to rebuild+start"); - return submit_and_emit( - coord, - templates::stale_start(agent, source, format!("{reason} (stale — rebuild+start)")), - ); + let mut targets = Vec::with_capacity(agents.len()); + for agent in agents { + targets.push((agent.clone(), lifecycle::is_running(agent).await)); } - submit_and_emit(coord, templates::start(agent, source, reason)) + submit_and_emit(coord, restart_spec(&targets, graceful, source, reason)) } -/// Hard stop: `SetWanted(Offline)` (a DAG node now) then reconcile (kill + -/// unregister + `Killed` event). -pub fn stop(coord: &Arc, agent: &str, source: Source, reason: String) -> u64 { - submit_and_emit(coord, templates::stop(agent, source, reason)) +/// Start a single agent. Thin wrapper over [`start_many`]. +pub async fn start(coord: &Arc, agent: &str, source: Source, reason: String) -> u64 { + start_many(coord, &[agent.to_owned()], source, reason).await } -/// Graceful stop: signal → drain → reconcile (the actual stop). The head -/// `SetWanted(Offline)` node writes the intent as part of the DAG. -pub fn graceful_stop(coord: &Arc, agent: &str, source: Source, reason: String) -> u64 { - submit_and_emit(coord, templates::graceful_stop(agent, source, reason)) +/// Start `agents` (one or many) in a **single** DAG — one per-agent subgraph +/// each, built dynamically from live state and run concurrently on their own +/// leases. A down agent gets `SetWanted(Up) → Reconcile` (or, rev stale, a +/// rebuild-then-start so it comes up on current derivations); an already- +/// running agent gets `SetWanted(Up) → Reconcile` (the reconcile noops). The +/// whole hive-wide `hivectl start` is one DAG. +pub async fn start_many( + coord: &Arc, + agents: &[String], + source: Source, + reason: String, +) -> u64 { + let current = crate::auto_update::current_flake_rev(&coord.hyperhive_flake); + let mut targets = Vec::with_capacity(agents.len()); + for agent in agents { + let running = lifecycle::is_running(agent).await; + let stored = std::fs::read_to_string(crate::paths::applied_rev_marker(agent)).ok(); + let stale = current + .as_ref() + .is_some_and(|rev| stored.as_deref() != Some(rev.as_str())); + if !running && stale { + tracing::info!(%agent, "start: rev stale + agent down — rebuild-then-start"); + } + targets.push((agent.clone(), running, stale)); + } + submit_and_emit(coord, start_spec(&targets, source, reason)) } -/// Graceful restart of a single agent: signal → drain → mechanical stop → -/// reconcile (starts it back up) — one atomic DAG, no client-side "await -/// the stop DAG then submit a start DAG" split. The head `SetWanted(Up)` -/// node writes the intent as part of the DAG. Thin wrapper over -/// [`restart_many`] with `graceful = true`. -pub fn graceful_restart( +/// Hard stop a single agent. Thin wrapper over [`stop_many`]. +pub async fn stop(coord: &Arc, agent: &str, source: Source, reason: String) -> u64 { + stop_many(coord, &[agent.to_owned()], false, source, reason).await +} + +/// Graceful stop of a single agent (signal → drain → reconcile, when +/// running). Thin wrapper over [`stop_many`] with `graceful = true`. +pub async fn graceful_stop( coord: &Arc, agent: &str, source: Source, reason: String, ) -> u64 { - restart_many(coord, &[agent.to_owned()], true, source, reason) + stop_many(coord, &[agent.to_owned()], true, source, reason).await +} + +/// Stop `agents` (one or many) in a **single** DAG — one per-agent subgraph +/// each, built dynamically from live state and run concurrently on their own +/// leases. A running agent gets `SetWanted(Off) → [Signal → Drain →](graceful) +/// Reconcile`; a down agent gets just `SetWanted(Off) → Reconcile` (skips the +/// pointless quiesce, keeps the Reconcile as the race-up backstop). The whole +/// hive-wide `hivectl stop` is one DAG. +pub async fn stop_many( + coord: &Arc, + agents: &[String], + graceful: bool, + source: Source, + reason: String, +) -> u64 { + let mut targets = Vec::with_capacity(agents.len()); + for agent in agents { + targets.push((agent.clone(), lifecycle::is_running(agent).await)); + } + submit_and_emit(coord, stop_spec(&targets, graceful, source, reason)) } /// Perm change: commit the JSON file(s) then rebuild. diff --git a/hive-c0re/src/job_queue/templates.rs b/hive-c0re/src/job_queue/templates.rs index be258043..f9a7bebd 100644 --- a/hive-c0re/src/job_queue/templates.rs +++ b/hive-c0re/src/job_queue/templates.rs @@ -4,47 +4,45 @@ //! `Vec` + `deps`). //! //! Every node carries its own `agent` (there is no DAG-level agent) — the -//! `node` helper stamps the template's agent onto each. `restart` takes an -//! agent *list* and stamps each agent onto its own subgraph, so a hive-wide -//! `hivectl restart` is ONE DAG with N independent per-agent subgraphs -//! (each a root chain, run concurrently on its own lease) rather than N -//! separate DAGs. The other templates are still single-agent. -//! -//! The power ops write the durable `wanted` intent via a head -//! `SetWanted(w)` node (not a pre-submit side effect); it holds the agent -//! lease so intent+reconcile is atomic per-agent. +//! `node` helper stamps each node's agent. This module holds the *pure* +//! shape builders (no I/O). The hive-wide **power ops** (`stop` / `start` / +//! `restart`) are NOT here: their per-agent shape depends on each agent's +//! live running state (an async `lifecycle::is_running` read), so they are +//! assembled dynamically in `submit.rs` out of the shared pure primitives +//! this module exports (`node`, `after_ok`, `rebuild_nodes`) — one +//! independent per-agent subgraph each, concurrent on its own lease, ONE +//! DAG for the whole hive-wide op. Power ops write the durable `wanted` +//! intent via a head `SetWanted(w)` node (holding the agent lease, so +//! intent+reconcile is atomic per-agent). //! //! ```text //! rebuild(a): Prebuild(a) → StopForUpdate(a) → Swap(a) →(any) Reconcile(a) -//! graceful-stop(a): SetWanted(a,Off) → Signal(a) → Drain(a) → Reconcile(a) -//! restart(a): SetWanted(a,Up) → StopForUpdate(a) → Reconcile(a) -//! graceful-restart(a): SetWanted(a,Up) → Signal(a) → Drain(a) → StopForUpdate(a) → Reconcile(a) -//! start(a): SetWanted(a,Up) → Reconcile(a) -//! stop(a): SetWanted(a,Off) → Reconcile(a) -//! stale-start(a): SetWanted(a,Up) → «rebuild subgraph» (rev stale; prebuild noops, agent down) //! spawn(a): Provision(a) → Create(a) → WriteDropin(a) → Reconcile(a) [wanted=Up at approve] //! perm-change(a): WritePermFile(a) → «rebuild subgraph» //! meta-update(inp): MetaLock(inp) → «fan-out rebuild(a) per affected a» -//! startup sweep: MetaLock(hyperhive, non-fatal) → «fan-out rebuild(stale a)» //! ``` +//! +//! For the dynamic power-op shapes (`stop` / `start` / `restart`, built from +//! live online/offline state), see `submit.rs`. use anyhow::{Result, bail}; use super::model::{DagSpec, Dep, DepWhen, NodeKind, NodeSpec, PermPayload, Source, Template}; use crate::coordinator::TransientKind; -/// After-ok edge on the previous node — the common chain link. -fn after_ok(on: u32) -> Vec { +/// After-ok edge on the previous node — the common chain link. Shared with +/// the async power-op builders in `submit.rs` (which assemble per-agent +/// chains dynamically from live container state). +pub(crate) fn after_ok(on: u32) -> Vec { vec![Dep { on, when: DepWhen::AfterOk, }] } -/// Build one node targeting `agent`. The single place templates stamp a -/// node's agent, so a whole template is single-agent by passing the same -/// `agent` to every call. -fn node(agent: &str, kind: NodeKind, deps: Vec) -> NodeSpec { +/// Build one node targeting `agent`. The single place a node's agent is +/// stamped. Shared with `submit.rs`'s dynamic power-op builders. +pub(crate) fn node(agent: &str, kind: NodeKind, deps: Vec) -> NodeSpec { NodeSpec { agent: agent.to_owned(), kind, @@ -56,7 +54,7 @@ fn node(agent: &str, kind: NodeKind, deps: Vec) -> NodeSpec { /// it must run even when the profile swap failed, so a previously-up /// agent comes back on its old config (today's recovery-start). This /// is the only `AfterAny` edge in v1. -fn rebuild_nodes(agent: &str, relock: bool, base: u32) -> Vec { +pub(crate) fn rebuild_nodes(agent: &str, relock: bool, base: u32) -> Vec { vec![ node( agent, @@ -122,73 +120,6 @@ pub fn approval_deploy(agent: &str, approval_id: i64, reason: String) -> DagSpec } } -/// Graceful stop: cheap `Signal` fires immediately (no build slot), the -/// `Drain` awaits the harness checkpoint (bounded), and the tail -/// `Reconcile` performs the actual container stop. The head `SetWanted` -/// node writes `wanted = Offline` as part of the DAG (was a pre-submit -/// side effect). A whole-hive graceful stop therefore signals every agent -/// up front and overlaps every drain, replacing the old detached-watcher -/// thread structurally. -pub fn graceful_stop(agent: &str, source: Source, reason: String) -> DagSpec { - DagSpec { - template: Template::GracefulStop, - source, - reason, - parent_id: None, - approval_id: None, - inputs: Vec::new(), - perm_payload: None, - transient: Some(TransientKind::Stopping), - nodes: vec![ - node(agent, NodeKind::SetWanted { up: false }, Vec::new()), - node(agent, NodeKind::Signal, after_ok(0)), - node(agent, NodeKind::Drain, after_ok(1)), - node(agent, NodeKind::Reconcile, after_ok(2)), - ], - } -} - -/// Restart one or more agents in a **single** DAG. Each agent gets an -/// independent subgraph — a head `SetWanted(Up)` (a root: no cross-agent -/// dep) then its restart chain — so all agents' restarts run concurrently, -/// each taking its own agent lease. `graceful` inserts `Signal → Drain` -/// before the mechanical `StopForUpdate` (each agent's subgraph is still one -/// atomic restart). One agent = the ordinary single-agent restart; many = -/// a hive-wide `hivectl restart` as one DAG instead of N separate ones. -pub fn restart(agents: &[String], graceful: bool, source: Source, reason: String) -> DagSpec { - let mut nodes = Vec::new(); - for agent in agents { - let base = u32::try_from(nodes.len()).unwrap_or(u32::MAX); - // Head of this agent's subgraph — a root (empty deps), so the N - // per-agent subgraphs are independent and run concurrently. - nodes.push(node(agent, NodeKind::SetWanted { up: true }, Vec::new())); - if graceful { - nodes.push(node(agent, NodeKind::Signal, after_ok(base))); - nodes.push(node(agent, NodeKind::Drain, after_ok(base + 1))); - nodes.push(node(agent, NodeKind::StopForUpdate, after_ok(base + 2))); - nodes.push(node(agent, NodeKind::Reconcile, after_ok(base + 3))); - } else { - nodes.push(node(agent, NodeKind::StopForUpdate, after_ok(base))); - nodes.push(node(agent, NodeKind::Reconcile, after_ok(base + 1))); - } - } - DagSpec { - template: if graceful { - Template::GracefulRestart - } else { - Template::Restart - }, - source, - reason, - parent_id: None, - approval_id: None, - inputs: Vec::new(), - perm_payload: None, - transient: Some(TransientKind::Restarting), - nodes, - } -} - /// Boot-time reconcile: a single `Reconcile` node that converges observed /// power state to the persisted intent — `wanted` is untouched (no /// `SetWanted`), unlike the operator `start`/`stop` templates. Used only @@ -213,68 +144,6 @@ pub fn reconcile_only( } } -/// Start: write `wanted = Up` (head `SetWanted`), then reconcile (which -/// starts the container). The intent write is a DAG node now, not a -/// pre-submit side effect. -pub fn start(agent: &str, source: Source, reason: String) -> DagSpec { - DagSpec { - template: Template::Start, - source, - reason, - parent_id: None, - approval_id: None, - inputs: Vec::new(), - perm_payload: None, - transient: Some(TransientKind::Starting), - nodes: vec![ - node(agent, NodeKind::SetWanted { up: true }, Vec::new()), - node(agent, NodeKind::Reconcile, after_ok(0)), - ], - } -} - -/// Stop: write `wanted = Offline` (head `SetWanted`), then reconcile -/// (kill + unregister + `Killed` event). -pub fn stop(agent: &str, source: Source, reason: String) -> DagSpec { - DagSpec { - template: Template::Stop, - source, - reason, - parent_id: None, - approval_id: None, - inputs: Vec::new(), - perm_payload: None, - transient: Some(TransientKind::Stopping), - nodes: vec![ - node(agent, NodeKind::SetWanted { up: false }, Vec::new()), - node(agent, NodeKind::Reconcile, after_ok(0)), - ], - } -} - -/// Stale start: a `start` whose rev marker is stale, so it rebuilds -/// before coming up — `SetWanted(Up)` → «rebuild subgraph» → `Reconcile` -/// (the rebuild's tail `Reconcile` starts it, since `wanted = Up`). The -/// rebuild nodes are the same `rebuild_nodes` chain a manual rebuild uses -/// (reused, not a variant); only the leading `SetWanted(Up)` intent -/// differs. Shows as a `Rebuild` on the dashboard like the old -/// submit-time upgrade did. -pub fn stale_start(agent: &str, source: Source, reason: String) -> DagSpec { - let mut nodes = vec![node(agent, NodeKind::SetWanted { up: true }, Vec::new())]; - nodes.extend(rebuild_nodes(agent, true, 1)); - DagSpec { - template: Template::Rebuild, - source, - reason, - parent_id: None, - approval_id: None, - inputs: Vec::new(), - perm_payload: None, - transient: Some(TransientKind::Rebuilding), - nodes, - } -} - /// First-deploy spawn (approval-driven): `Provision` (proposed/applied /// repos, state subvolume, meta registration) then `Create` /// (`nixos-container create`), drop-in write, then `Reconcile` starts diff --git a/hive-c0re/src/job_queue/tests.rs b/hive-c0re/src/job_queue/tests.rs index 9a691ce8..a006f1a4 100644 --- a/hive-c0re/src/job_queue/tests.rs +++ b/hive-c0re/src/job_queue/tests.rs @@ -16,6 +16,22 @@ fn rebuild(agent: &str, reason: &str) -> DagSpec { templates::rebuild(agent, Source::Manual, reason.to_owned(), None, true) } +/// Restart DAG spec with every agent treated as **running** — the online +/// shape (`SetWanted → [Signal→Drain→] StopForUpdate → Reconcile`) most +/// queue-mechanics tests assume. Mirrors the pre-dynamic `templates::restart` +/// (which is now the state-aware `submit::restart_spec`). +fn restart_online(agents: &[&str], graceful: bool, reason: &str) -> DagSpec { + let targets: Vec<(String, bool)> = agents.iter().map(|a| ((*a).to_owned(), true)).collect(); + submit::restart_spec(&targets, graceful, Source::Manual, reason.to_owned()) +} + +/// Stop DAG spec with every agent treated as **running** — the online shape +/// (`SetWanted → [Signal→Drain→](graceful) Reconcile`). +fn stop_online(agents: &[&str], graceful: bool, reason: &str) -> DagSpec { + let targets: Vec<(String, bool)> = agents.iter().map(|a| ((*a).to_owned(), true)).collect(); + submit::stop_spec(&targets, graceful, Source::Manual, reason.to_owned()) +} + /// Claim helper asserting exactly one node comes back. fn claim_one(q: &JobQueue) -> Claim { let mut claims = q.claim_ready(); @@ -65,15 +81,7 @@ fn distinct_submits_never_collapse() { let q = JobQueue::new(1); let a = submit(&q, rebuild("agent-a", "r")); let b = submit(&q, rebuild("agent-b", "r")); - let c = submit( - &q, - templates::restart( - &["agent-a".to_owned()], - false, - Source::Manual, - "r".to_owned(), - ), - ); + let c = submit(&q, restart_online(&["agent-a"], false, "r")); assert_ne!(a, b); assert_ne!(a, c); assert_eq!(q.snapshot().len(), 3); @@ -203,15 +211,7 @@ fn fifo_fairness_for_the_slot() { #[test] fn lease_serializes_two_lifecycle_dags_for_same_agent() { let q = JobQueue::new(4); - let restart = submit( - &q, - templates::restart( - &["agent-a".to_owned()], - false, - Source::Manual, - "restart".to_owned(), - ), - ); + let restart = submit(&q, restart_online(&["agent-a"], false, "restart")); let stop = submit( &q, templates::reconcile_only( @@ -292,24 +292,8 @@ fn lease_exempt_prebuild_overlaps_other_dag_on_same_agent() { #[test] fn agents_do_not_contend_on_each_others_leases() { let q = JobQueue::new(4); - submit( - &q, - templates::restart( - &["agent-a".to_owned()], - false, - Source::Manual, - "r".to_owned(), - ), - ); - submit( - &q, - templates::restart( - &["agent-b".to_owned()], - false, - Source::Manual, - "r".to_owned(), - ), - ); + submit(&q, restart_online(&["agent-a"], false, "r")); + submit(&q, restart_online(&["agent-b"], false, "r")); let claims = q.claim_ready(); assert_eq!(claims.len(), 2, "different agents run concurrently"); } @@ -319,12 +303,7 @@ fn multi_agent_restart_is_one_dag_with_concurrent_per_agent_subgraphs() { let q = JobQueue::new(4); let id = submit( &q, - templates::restart( - &["agent-a".to_owned(), "agent-b".to_owned()], - false, - Source::Manual, - "hive-wide".to_owned(), - ), + restart_online(&["agent-a", "agent-b"], false, "hive-wide"), ); // A hive-wide restart is ONE DAG, not one-per-agent. assert_eq!(q.snapshot().len(), 1); @@ -348,6 +327,129 @@ fn multi_agent_restart_is_one_dag_with_concurrent_per_agent_subgraphs() { ); } +#[test] +fn multi_agent_stop_is_one_dag_with_concurrent_per_agent_subgraphs() { + let q = JobQueue::new(4); + let id = submit( + &q, + stop_online(&["agent-a", "agent-b"], false, "hive-wide stop"), + ); + // A hive-wide stop is ONE DAG, not one-per-agent. + assert_eq!(q.snapshot().len(), 1); + let claims = q.claim_ready(); + assert!(claims.iter().all(|c| c.dag_id == id)); + let mut heads: Vec<(&str, &str, bool)> = claims + .iter() + .map(|c| (c.agent.as_str(), c.kind.as_str(), c.lease_acquired)) + .collect(); + heads.sort_unstable(); + assert_eq!( + heads, + vec![ + ("agent-a", "set_wanted", true), + ("agent-b", "set_wanted", true), + ], + "both per-agent stop subgraphs start concurrently, each on its own lease" + ); +} + +#[test] +fn multi_agent_start_one_dag_folds_per_agent_stale_rebuild() { + let q = JobQueue::new(4); + let id = submit( + &q, + // fresh: offline + not stale → SetWanted → Reconcile. + // stale: offline + stale → SetWanted → «rebuild subgraph». + submit::start_spec( + &[ + ("fresh".to_owned(), false, false), + ("stale".to_owned(), false, true), + ], + Source::Manual, + "hive-wide start".to_owned(), + ), + ); + // One DAG spanning both agents. + assert_eq!(q.snapshot().len(), 1); + // Both subgraph heads (SetWanted(Up)) are roots — claimable at once, + // each acquiring its own agent lease. + let heads = q.claim_ready(); + assert!( + heads + .iter() + .all(|c| c.dag_id == id && c.kind.as_str() == "set_wanted") + ); + let mut head_agents: Vec<&str> = heads.iter().map(|c| c.agent.as_str()).collect(); + head_agents.sort_unstable(); + assert_eq!(head_agents, vec!["fresh", "stale"]); + // Complete both heads; the fresh agent then reconciles directly while + // the stale agent's subgraph is the rebuild chain (prebuild first). + for c in &heads { + q.complete_node(id, c.node_id, Ok(())); + } + let next = q.claim_ready(); + let mut kinds: Vec<(&str, &str)> = next + .iter() + .map(|c| (c.agent.as_str(), c.kind.as_str())) + .collect(); + kinds.sort_unstable(); + assert_eq!( + kinds, + vec![("fresh", "reconcile"), ("stale", "prebuild")], + "fresh agent starts directly; stale agent rebuilds first, all in one DAG" + ); +} + +#[test] +fn offline_agents_skip_mechanical_nodes_but_keep_reconcile() { + // The dynamic build skips Signal/Drain/StopForUpdate for a down agent + // (nothing to quiesce/stop) but ALWAYS keeps the Reconcile tail — the + // convergence guarantee that catches a race-up between the is_running + // read and node exec. + let q = JobQueue::new(4); + // Offline graceful stop → SetWanted(Off) → Reconcile (no Signal/Drain). + let stop = submit( + &q, + submit::stop_spec( + &[("down".to_owned(), false)], + true, + Source::Manual, + "stop down".to_owned(), + ), + ); + // Offline restart → SetWanted(Up) → Reconcile (no StopForUpdate): a + // restart of a down agent is really a start. + let restart = submit( + &q, + submit::restart_spec( + &[("down2".to_owned(), false)], + true, + Source::Manual, + "restart down".to_owned(), + ), + ); + let shape = |id: u64| -> Vec { + q.snapshot() + .iter() + .find(|d| d.id == id) + .expect("dag") + .nodes + .iter() + .map(|n| n.kind.clone()) + .collect() + }; + assert_eq!( + shape(stop), + vec!["set_wanted".to_owned(), "reconcile".to_owned()], + "offline graceful stop skips the signal/drain quiesce, keeps Reconcile" + ); + assert_eq!( + shape(restart), + vec!["set_wanted".to_owned(), "reconcile".to_owned()], + "offline restart skips StopForUpdate, keeps Reconcile (it's a start)" + ); +} + // ---- failure: cancel-downstream + AfterAny ---- #[test] @@ -566,15 +668,7 @@ fn append_children_sets_parent() { #[test] fn terminal_dag_reported_exactly_once_and_lease_released() { let q = JobQueue::new(1); - let id = submit( - &q, - templates::restart( - &["agent-a".to_owned()], - false, - Source::Manual, - "r".to_owned(), - ), - ); + let id = submit(&q, restart_online(&["agent-a"], false, "r")); // restart = SetWanted → StopForUpdate → Reconcile; not terminal until // the last node completes. let set_wanted = claim_one(&q); @@ -681,8 +775,8 @@ fn trim_keeps_terminal_parent_with_live_children() { ); let lock = claim_one(&q); q.complete_node(meta, lock.node_id, Ok(())); - let mut child_spec = templates::restart( - &["agent-x".to_owned()], + let mut child_spec = submit::restart_spec( + &[("agent-x".to_owned(), true)], false, Source::MetaUpdate, "cascade".to_owned(), @@ -798,10 +892,7 @@ fn error_is_truncated() { #[test] fn graceful_stop_shape_signal_drain_reconcile() { let q = JobQueue::new(1); - let id = submit( - &q, - templates::graceful_stop("agent-a", Source::Manual, "graceful".to_owned()), - ); + let id = submit(&q, stop_online(&["agent-a"], true, "graceful")); for expected in ["set_wanted", "signal", "drain", "reconcile"] { let c = claim_one(&q); assert_eq!(c.kind.as_str(), expected); @@ -816,14 +907,8 @@ fn graceful_signal_and_drain_hold_no_build_slot() { // buildSlots = 1 while a rebuild hogs the slot. let q = JobQueue::new(1); submit(&q, rebuild("builder", "slot hog")); - submit( - &q, - templates::graceful_stop("agent-a", Source::Manual, "g".to_owned()), - ); - submit( - &q, - templates::graceful_stop("agent-b", Source::Manual, "g".to_owned()), - ); + submit(&q, stop_online(&["agent-a"], true, "g")); + submit(&q, stop_online(&["agent-b"], true, "g")); let claims = q.claim_ready(); let kinds: Vec<&str> = claims.iter().map(|c| c.kind.as_str()).collect(); assert_eq!( diff --git a/hive-c0re/src/server.rs b/hive-c0re/src/server.rs index d2d4d27c..0ba844cf 100644 --- a/hive-c0re/src/server.rs +++ b/hive-c0re/src/server.rs @@ -97,8 +97,8 @@ async fn dispatch(req: &HostRequest, coord: Arc) -> HostResponse { tracing::info!(%id, %name, "spawn approval queued"); HostResponse::success() } - HostRequest::Kill { name } => submit_single(&coord, name, Verb::Kill), - HostRequest::Restart { name } => submit_single(&coord, name, Verb::Restart), + HostRequest::Kill { name } => submit_single(&coord, name, Verb::Kill).await, + HostRequest::Restart { name } => submit_single(&coord, name, Verb::Restart).await, HostRequest::RestartAll => handle_restart_all(&coord).await?, HostRequest::RestartScoped { scope, graceful } => { handle_restart_scoped(&coord, scope, *graceful).await? @@ -143,7 +143,7 @@ async fn dispatch(req: &HostRequest, coord: Arc) -> HostResponse { actions::destroy(&coord, name, *purge).await?; HostResponse::success() } - HostRequest::Rebuild { name } => submit_single(&coord, name, Verb::Rebuild), + HostRequest::Rebuild { name } => submit_single(&coord, name, Verb::Rebuild).await, HostRequest::QueueDag { id } => { // The polled DAG first, then its live fan-out children. let dags = coord @@ -558,7 +558,7 @@ enum Verb { Rebuild, } -fn submit_single(coord: &Arc, name: &str, verb: Verb) -> HostResponse { +async fn submit_single(coord: &Arc, name: &str, verb: Verb) -> HostResponse { use crate::job_queue::{Source, submit}; let id = match verb { Verb::Kill => { @@ -569,6 +569,7 @@ fn submit_single(coord: &Arc, name: &str, verb: Verb) -> HostRespon Source::Manual, "manual kill via hivectl".to_owned(), ) + .await } Verb::Restart => { tracing::info!(%name, "restart"); @@ -578,6 +579,7 @@ fn submit_single(coord: &Arc, name: &str, verb: Verb) -> HostRespon Source::Manual, "manual restart via hivectl".to_owned(), ) + .await } Verb::Rebuild => { tracing::info!(%name, "rebuild"); @@ -606,13 +608,16 @@ async fn handle_restart_all(coord: &Arc) -> Result { let queued = if agents.is_empty() { Vec::new() } else { - vec![crate::job_queue::submit::restart_many( - coord, - &agents, - false, - crate::job_queue::Source::Manual, - "manual restart via hivectl restart-all".to_owned(), - )] + vec![ + crate::job_queue::submit::restart_many( + coord, + &agents, + false, + crate::job_queue::Source::Manual, + "manual restart via hivectl restart-all".to_owned(), + ) + .await, + ] }; let mut resp = HostResponse::list(agents); resp.queued_dags = Some(queued); @@ -644,29 +649,27 @@ async fn handle_stop( let mut errors: Vec = Vec::new(); let mut queued: Vec = Vec::new(); - for agent in agents { + // One DAG for all targeted agents — a per-agent stop subgraph each + // (`SetWanted(Offline) → [Signal → Drain →] Reconcile`), independent + // roots that run concurrently on their own leases. A hive-wide + // `hivectl stop` is now a single DAG, not N. + if !agents.is_empty() { let reason = if graceful { "manual via hivectl graceful stop" } else { "manual via hivectl stop" }; - let id = if graceful { - crate::job_queue::submit::graceful_stop( + queued.push( + crate::job_queue::submit::stop_many( coord, - agent, + agents, + graceful, crate::job_queue::Source::Manual, 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()); + .await, + ); + ok_items.extend(agents.iter().cloned()); } // Agents go down before infra so they're not mid-request against a @@ -739,18 +742,25 @@ async fn handle_start( } } + // One DAG for all targeted agents — a per-agent start subgraph each + // (`SetWanted(Up) → Reconcile`, or a rebuild-then-start for a stale + // rev), independent roots that run concurrently on their own leases. A + // hive-wide `hivectl start` is now a single DAG, not N. Through the + // queue: persists `wanted = Up`, per-agent stale-rev upgrade to a full + // rebuild, serializes on each agent's lease. The id rides back for + // hivectl's wait loop. let mut queued: Vec = Vec::new(); - for agent in agents { - // 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()); + if !agents.is_empty() { + queued.push( + crate::job_queue::submit::start_many( + coord, + agents, + crate::job_queue::Source::Manual, + "manual via hivectl start".to_owned(), + ) + .await, + ); + ok_items.extend(agents.iter().cloned()); } let mut resp = finish_lifecycle(ok_items, &errors); @@ -786,17 +796,20 @@ async fn handle_restart_scoped( // client-side stop-then-start composition — the whole restart survives // a dropped connection because the DAG owns it. if !agents.is_empty() { - queued.push(crate::job_queue::submit::restart_many( - coord, - &agents, - graceful, - crate::job_queue::Source::Manual, - if graceful { - "manual via hivectl restart --graceful".to_owned() - } else { - "manual restart via hivectl restart".to_owned() - }, - )); + queued.push( + crate::job_queue::submit::restart_many( + coord, + &agents, + graceful, + crate::job_queue::Source::Manual, + if graceful { + "manual via hivectl restart --graceful".to_owned() + } else { + "manual restart via hivectl restart".to_owned() + }, + ) + .await, + ); ok_items.extend(agents.iter().cloned()); } diff --git a/hive-c0re/src/socket_server/lifecycle_handlers.rs b/hive-c0re/src/socket_server/lifecycle_handlers.rs index 86f24ca3..c0ef4d20 100644 --- a/hive-c0re/src/socket_server/lifecycle_handlers.rs +++ b/hive-c0re/src/socket_server/lifecycle_handlers.rs @@ -12,7 +12,11 @@ use crate::coordinator::Coordinator; /// `Start` — start a container, kicking its next turn. The caller must be an /// ancestor of `name` in the topology (the root covers every agent). -pub(super) fn handle_start(coord: &Arc, agent: &str, name: &str) -> AgentResponse { +pub(super) async fn handle_start( + coord: &Arc, + agent: &str, + name: &str, +) -> AgentResponse { if let Some(err) = require_descendant(agent, name, "start") { return err; } @@ -25,7 +29,8 @@ pub(super) fn handle_start(coord: &Arc, agent: &str, name: &str) -> name, crate::job_queue::Source::Manual, format!("agent `{agent}` start tool"), - ); + ) + .await; AgentResponse::Ok } @@ -56,7 +61,8 @@ pub(super) async fn handle_restart( name, crate::job_queue::Source::Manual, format!("agent `{agent}` restart tool"), - ); + ) + .await; AgentResponse::Ok } diff --git a/hive-c0re/src/socket_server/mod.rs b/hive-c0re/src/socket_server/mod.rs index 1f00baa7..27a2ec15 100644 --- a/hive-c0re/src/socket_server/mod.rs +++ b/hive-c0re/src/socket_server/mod.rs @@ -557,7 +557,7 @@ async fn dispatch(req: &AgentRequest, agent: &str, coord: &Arc) -> match req { // Lifecycle + config: caller must be an ancestor of the target // (a parent owns its whole subtree; the root covers every agent). - AgentRequest::Start { name } => handle_start(coord, agent, name), + AgentRequest::Start { name } => handle_start(coord, agent, name).await, AgentRequest::Restart { name } => handle_restart(coord, agent, name).await, AgentRequest::Kill { name } => handle_kill(coord, agent, name).await, AgentRequest::Update { name } => handle_update(coord, agent, name),