diff --git a/docs/coordinator.md b/docs/coordinator.md index e990d359..e915b0a4 100644 --- a/docs/coordinator.md +++ b/docs/coordinator.md @@ -40,8 +40,9 @@ somewhere." | `Destroy` | For future use (`destroy --purge` does real I/O). Variant exists so the wire shape doesn't change later; not currently routed through the queue. | | `Restart` | Stop + start a container without touching config (~5-10s). Routed through the queue so it serialises against in-flight rebuilds for the same agent — prevents a restart racing a rebuild mid-flight. Sources: dashboard ↺ button, manager `restart` MCP tool. | | `PermChange` | Write a tool-group or capability change to the shared JSON file (`tool-groups.json` / `capabilities.json`), then rebuild the agent so the updated `HIVE_TOOL_GROUPS` / `HIVE_CAPABILITIES` env var takes effect. Serialising the file write through the queue prevents concurrent dashboard batch-apply actions from racing on the shared file. After a successful file write, emits `CapabilitiesChanged` or `ToolGroupsChanged` SSE snapshot so the P3RM1SS10NS tab updates live. | +| `GracefulStop` | Quiesce then stop a container (the `?graceful=1` path on `/kill/`). Signals the harness (its next `Recv` returns `GracefulStop` — the inbound fence — so it runs one stop-checkpoint turn to flush durable `/state`, then exits), waits for it to drain (bounded by a 3-min timeout → hard-stop fallback), then runs the normal container-stop teardown. Queued so it can't race an in-flight rebuild for the same agent. | -**Intentionally not queued** (sub-second ops): `start`, `stop`, `kill`. +**Intentionally not queued** (sub-second ops): the *hard* `start`, `stop`, `kill`. (A *graceful* stop is the `GracefulStop` kind above — it takes a checkpoint turn, so it rides the queue.) ### Dedup diff --git a/hive-ag3nt/src/bin/hive.rs b/hive-ag3nt/src/bin/hive.rs index fe19e4f0..0697ddca 100644 --- a/hive-ag3nt/src/bin/hive.rs +++ b/hive-ag3nt/src/bin/hive.rs @@ -173,6 +173,22 @@ fn synthetic_continue() -> hive_sh4re::DeliveredMessage { } } +/// Synthetic message that drives the single stop-checkpoint turn when c0re +/// signals a graceful stop. The agent gets one final turn to flush durable +/// `/state` before the container is stopped; new inbound is already fenced. +fn graceful_stop_message() -> hive_sh4re::DeliveredMessage { + hive_sh4re::DeliveredMessage { + from: "graceful-stop".into(), + body: "You are being gracefully stopped — the container will shut down after this turn, \ + and new inbound messages are already fenced. Flush anything worth keeping to your \ + durable /state files now, then end your turn. Do not start new long-running work." + .into(), + id: 0, + redelivered: false, + in_reply_to: None, + } +} + // ---------- surface trait ---------- /// What a `Recv` long-poll returned. Decoupled from the per-role @@ -188,6 +204,10 @@ enum RecvOutcome { /// retries; the surface impl is responsible for tracing the /// detail before returning this. TransportError, + /// c0re signalled a graceful stop for this agent. The serve loop runs + /// one stop-checkpoint turn (flush durable `/state`), reports + /// `GracefulStopComplete`, and exits so the container can be stopped. + GracefulStop, } /// Wire surface abstraction. `AgentSurface` is the only impl — the trait @@ -211,6 +231,12 @@ trait Surface { /// Either field is `None` when the underlying request errors. fn post_turn_counts(socket: &Path) -> impl Future, Option)>; + /// Tell c0re the graceful-stop checkpoint is done and the harness is + /// exiting its serve loop (fire-and-forget; logs on error). Lets the + /// `GracefulStop` orchestration stop the container without waiting out + /// its timeout fallback. + fn graceful_stop_complete(socket: &Path) -> impl Future; + /// Send a message addressed to `` (broker resolves the /// sentinel via `topology::parent_of` at delivery time; root /// agents/manager fall through to operator). @@ -259,6 +285,18 @@ impl Surface for AgentSurface { } } + async fn graceful_stop_complete(socket: &Path) { + match client::request::<_, AgentResponse>(socket, &AgentRequest::GracefulStopComplete).await + { + Ok(AgentResponse::Ok) => {} + Ok(AgentResponse::Err { message }) => { + tracing::warn!(%message, "graceful_stop_complete rejected by broker"); + } + Ok(other) => tracing::warn!(?other, "graceful_stop_complete unexpected response"), + Err(e) => tracing::warn!(error = ?e, "graceful_stop_complete transport error"), + } + } + async fn inbox_unread(socket: &Path) -> u64 { match client::request::<_, AgentResponse>(socket, &AgentRequest::Status).await { Ok(AgentResponse::Status { unread }) => unread, @@ -318,6 +356,7 @@ impl Surface for AgentSurface { RecvOutcome::Message(first) } Ok(AgentResponse::Messages { .. }) => RecvOutcome::Empty, + Ok(AgentResponse::GracefulStop) => RecvOutcome::GracefulStop, Ok(AgentResponse::Err { message }) => { tracing::warn!(%message, "recv error"); RecvOutcome::TransportError @@ -473,6 +512,26 @@ async fn serve_loop( // No backoff: the long-poll wait is itself the throttle. continue; } + RecvOutcome::GracefulStop => { + // c0re fenced our inbox and wants a clean stop. Run one + // checkpoint turn so the agent flushes durable /state, + // report completion, then exit the loop → the harness + // process ends and the container can be stopped. + tracing::info!( + "graceful stop signalled — running stop-checkpoint turn, then exiting" + ); + let _ = handle_turn::( + socket, + &bus, + stats.as_ref(), + files, + &turn_lock, + graceful_stop_message(), + ) + .await; + S::graceful_stop_complete(socket).await; + return Ok(()); + } }, }; let ctrl = handle_turn::(socket, &bus, stats.as_ref(), files, &turn_lock, next).await; diff --git a/hive-ag3nt/src/mcp.rs b/hive-ag3nt/src/mcp.rs index 34082d72..e4ddebc0 100644 --- a/hive-ag3nt/src/mcp.rs +++ b/hive-ag3nt/src/mcp.rs @@ -79,6 +79,10 @@ impl From for SocketReply { hive_sh4re::Response::HostJournal { content } => Self::HostJournal(content), hive_sh4re::Response::Schedules { schedules } => Self::Schedules(schedules), hive_sh4re::Response::Containers { containers } => Self::Containers(containers), + // A graceful stop is pending — the inbox is fenced. If claude polls + // `recv` during its stop-checkpoint turn, present it as an empty + // inbox (the serve loop is the real handler of this signal). + hive_sh4re::Response::GracefulStop => Self::Messages(Vec::new()), hive_sh4re::Response::AgentMeta { name, running, diff --git a/hive-c0re/src/agent_server.rs b/hive-c0re/src/agent_server.rs index 2808fe70..f59af5ce 100644 --- a/hive-c0re/src/agent_server.rs +++ b/hive-c0re/src/agent_server.rs @@ -185,6 +185,13 @@ pub(crate) async fn dispatch_shared( } hive_sh4re::Request::AckTurn => handle_ack_turn(coord, agent), hive_sh4re::Request::RequeueInflight => handle_requeue_inflight(coord, agent), + hive_sh4re::Request::GracefulStopComplete => { + // Harness drained + is exiting: clear the fence so the + // `GracefulStop` orchestration (which polls this flag) proceeds + // to stop the container without waiting out its timeout. + coord.clear_graceful_stop(agent); + hive_sh4re::Response::Ok + } hive_sh4re::Request::GetHostJournal { unit, container, @@ -221,6 +228,15 @@ async fn handle_recv( wait_seconds: Option, max: Option, ) -> hive_sh4re::Response { + // Graceful-stop fence: while a graceful stop is pending for this agent, + // return `GracefulStop` instead of polling the broker. The harness runs + // one stop-checkpoint turn then exits; new sends keep queueing in the + // broker for the agent's next start. Checked before the (blocking) poll + // so a flag set between polls is seen on the next Recv — the orchestration + // also fires a transient wake to break an in-flight long-poll. + if coord.is_graceful_stop_pending(agent) { + return hive_sh4re::Response::GracefulStop; + } let cap = max.unwrap_or(1).min(RECV_BATCH_MAX) as usize; match coord .broker diff --git a/hive-c0re/src/coordinator.rs b/hive-c0re/src/coordinator.rs index cff91a2f..451a94a2 100644 --- a/hive-c0re/src/coordinator.rs +++ b/hive-c0re/src/coordinator.rs @@ -125,6 +125,12 @@ pub struct Coordinator { /// the dashboard's `agents_crashing` banner warning via /// `recent_crash_counts`, which prunes entries older than its window. recent_crashes: Mutex>>, + /// Agents with a graceful stop in progress. Set by the `GracefulStop` + /// orchestration; read by `agent_server::handle_recv`, which returns + /// `Response::GracefulStop` (instead of polling the broker) while an + /// agent is in this set — the inbound fence. Cleared when the agent + /// reports `GracefulStopComplete` or the container is stopped. + graceful_stop_pending: Mutex>, /// Unified wire-facing event channel feeding the dashboard SSE /// stream. Carries broker messages (mirrored from `broker.subscribe` /// by the forwarder task in `main.rs`) and dashboard-only mutation @@ -450,6 +456,7 @@ impl Coordinator { transient: Mutex::new(HashMap::new()), recent_transient: Mutex::new(HashMap::new()), recent_crashes: Mutex::new(HashMap::new()), + graceful_stop_pending: Mutex::new(HashSet::new()), dashboard_events, event_seq: AtomicU64::new(0), meta_updates_active: AtomicU64::new(0), @@ -1128,6 +1135,28 @@ impl Coordinator { } } + /// Mark `name` as having a graceful stop in progress. While set, + /// `agent_server::handle_recv` returns `Response::GracefulStop` for + /// this agent instead of polling the broker (the inbound fence). + pub fn mark_graceful_stop(&self, name: &str) { + self.graceful_stop_pending + .lock() + .unwrap() + .insert(name.to_owned()); + } + + /// Whether a graceful stop is pending for `name`. + #[must_use] + pub fn is_graceful_stop_pending(&self, name: &str) -> bool { + self.graceful_stop_pending.lock().unwrap().contains(name) + } + + /// Clear the graceful-stop flag for `name` (agent reported + /// `GracefulStopComplete`, or the stop finished / was abandoned). + pub fn clear_graceful_stop(&self, name: &str) { + self.graceful_stop_pending.lock().unwrap().remove(name); + } + /// Set of agents whose transient was cleared within the last /// `grace` seconds — i.e. agents the operator just acted on, /// whose stop the crash watcher should NOT classify as a crash. diff --git a/hive-c0re/src/dashboard/lifecycle_ops.rs b/hive-c0re/src/dashboard/lifecycle_ops.rs index 7bc7c0aa..f55cad61 100644 --- a/hive-c0re/src/dashboard/lifecycle_ops.rs +++ b/hive-c0re/src/dashboard/lifecycle_ops.rs @@ -6,12 +6,21 @@ //! spinner); destroy delegates to `actions::destroy` (optionally purging). use axum::{ - extract::{Form, Path as AxumPath, State}, + extract::{Form, Path as AxumPath, Query, State}, http::StatusCode, response::{IntoResponse, Response}, }; use serde::Deserialize; +/// Query params for `post_kill`. `?graceful=1` routes to the graceful-stop +/// orchestration (quiesce the harness, flush `/state`, then container stop) +/// instead of an immediate hard stop. Defaults false → today's hard kill. +#[derive(Deserialize)] +pub(super) struct KillParams { + #[serde(default)] + graceful: bool, +} + use super::{AppState, error_response, guard_agent_name, strip_container_prefix}; use crate::{actions, lifecycle}; @@ -75,11 +84,28 @@ where pub(super) async fn post_kill( State(state): State, AxumPath(name): AxumPath, + Query(params): Query, ) -> Response { let logical = strip_container_prefix(&name); if let Some(reject) = guard_agent_name(&state, &logical).await { return reject; } + if params.graceful { + // Graceful stop: enqueue the quiesce orchestration (signal the harness + // → one stop-checkpoint turn → drain → container stop, with a timeout + // fallback to a hard stop). Serialised through the rebuild queue so it + // can't race an in-flight rebuild for the same agent, and its per-step + // progress surfaces on the queue snapshot + build log. + state.coord.rebuild_queue.enqueue( + crate::rebuild_queue::QueueKind::GracefulStop, + logical, + crate::rebuild_queue::QueueSource::Manual, + "manual via dashboard graceful stop".to_owned(), + None, + ); + state.coord.emit_rebuild_queue_snapshot(); + return (StatusCode::OK, "ok").into_response(); + } // Manager is stoppable from the dashboard like any other // agent. The host's dashboard server keeps running (it's // hive-c0re, not the manager container), per-agent approvals diff --git a/hive-c0re/src/rebuild_queue.rs b/hive-c0re/src/rebuild_queue.rs index 8b503484..1c78aa1b 100644 --- a/hive-c0re/src/rebuild_queue.rs +++ b/hive-c0re/src/rebuild_queue.rs @@ -42,6 +42,10 @@ pub enum QueueKind { /// Serialised through the queue so concurrent dashboard batch-apply /// actions for different agents never race on the shared JSON file. PermChange, + /// Gracefully stop a container: signal the harness to run one + /// stop-checkpoint turn (flush durable `/state`), wait for it to drain, + /// then `nixos-container stop`. Falls back to a hard stop on timeout. + GracefulStop, } impl QueueKind { @@ -54,6 +58,7 @@ impl QueueKind { QueueKind::StartupSweep => "startup_sweep", QueueKind::Restart => "restart", QueueKind::PermChange => "perm_change", + QueueKind::GracefulStop => "graceful_stop", } } } @@ -662,6 +667,12 @@ impl RebuildQueue { /// signal the worker exits after its current entry finishes; pending /// `Queued` entries are dropped (they'll either be replayed by the /// startup sweep on next boot or left for an operator to re-queue). +/// Max time the `GracefulStop` worker waits for the harness to run its +/// stop-checkpoint turn + drain before falling back to a hard container stop. +/// Generous — a checkpoint turn can take a while — but bounded so a wedged +/// agent never blocks the stop indefinitely. +const GRACEFUL_STOP_TIMEOUT: std::time::Duration = std::time::Duration::from_mins(3); + pub async fn run_worker(coord: std::sync::Arc) { let mut shutdown = coord.shutdown_rx(); loop { @@ -833,9 +844,49 @@ async fn dispatch( crate::auto_update::current_flake_rev(&coord.hyperhive_flake).unwrap_or_default(); crate::auto_update::rebuild_agent(coord, name, ¤t_rev, Some(entry.id)).await } + (QueueKind::GracefulStop, _) => run_graceful_stop(coord, entry).await, } } +/// Run one `GracefulStop` entry: signal the harness to quiesce (it returns +/// `GracefulStop` on its next `Recv`, runs one stop-checkpoint turn to flush +/// durable `/state`, then exits), wait for it to drain — bounded by +/// `GRACEFUL_STOP_TIMEOUT` so a wedged agent can't block forever — then stop +/// the container with the same teardown as a plain kill. +async fn run_graceful_stop( + coord: &std::sync::Arc, + entry: &QueueEntry, +) -> anyhow::Result<()> { + let name = &entry.agent; + let _guard = coord.transient_guard(name, crate::coordinator::TransientKind::Stopping); + // Signal the harness; the kick breaks an idle long-poll so it's seen promptly. + coord.set_queue_step(Some(entry.id), "graceful stop: signalling agent"); + coord.mark_graceful_stop(name); + coord.kick_agent(name, "graceful stop requested"); + // Wait for the harness to drain (it clears the flag via `GracefulStopComplete`) + // or fall back to a hard stop after the timeout. The single queue worker is + // intentionally held for the duration — graceful stops are infrequent. + coord.set_queue_step(Some(entry.id), "graceful stop: waiting for agent to drain"); + let deadline = std::time::Instant::now() + GRACEFUL_STOP_TIMEOUT; + while coord.is_graceful_stop_pending(name) { + if std::time::Instant::now() >= deadline { + tracing::warn!(agent = %name, "graceful stop: drain timed out — hard-stopping"); + break; + } + tokio::time::sleep(std::time::Duration::from_millis(500)).await; + } + coord.clear_graceful_stop(name); + // Stop the container — same teardown as a plain kill. + coord.set_queue_step(Some(entry.id), "nixos-container stop"); + crate::lifecycle::kill(name).await?; + coord.unregister_agent(name); + coord.notify_manager(&hive_sh4re::HelperEvent::Killed { + agent: name.clone(), + }); + coord.rescan_containers_and_emit().await; + Ok(()) +} + /// Run one `MetaUpdate` entry: bump the meta flake's locks for the /// requested inputs, then enqueue a cascade of `Rebuild` entries /// (with `parent_id` set to this entry's id) for every agent affected diff --git a/hive-sh4re/src/lib.rs b/hive-sh4re/src/lib.rs index ba3789a7..73f7234c 100644 --- a/hive-sh4re/src/lib.rs +++ b/hive-sh4re/src/lib.rs @@ -529,6 +529,11 @@ pub enum Request { /// crashed-mid-turn sessions. See /// `docs/conventions.md::Broker delivery + ack cycle`. RequeueInflight, + /// Harness → c0re: "I saw the `GracefulStop` signal, ran my + /// stop-checkpoint turn (durable `/state` flushed) and am exiting my + /// serve loop now." Lets the `GracefulStop` orchestration stop the + /// container immediately instead of waiting out its timeout fallback. + GracefulStopComplete, /// *(capability-gated: `read_host_journal`)* Fetch recent lines /// from the host journal. Filters are all optional; omitting all /// returns the last `lines` entries from the global journal. @@ -700,6 +705,15 @@ pub enum Response { /// status. Ordered by topology depth (parents before children), then /// alphabetically within each depth tier. Containers { containers: Vec }, + /// `Recv` result when a graceful stop is pending for this agent + /// (set by hive-c0re's `GracefulStop` orchestration). Returned in + /// place of `Messages` — it doubles as the inbound fence: the harness + /// stops consuming normal inbox messages and instead runs one + /// stop-checkpoint turn (flush durable `/state`), then reports + /// `GracefulStopComplete` and exits its serve loop so the container + /// can be stopped cleanly. New sends keep queueing in the broker for + /// the agent's next start. + GracefulStop, } /// Backwards-compatible response aliases.