diff --git a/Cargo.lock b/Cargo.lock index 693b1121..bcab7b39 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1795,7 +1795,6 @@ dependencies = [ "anyhow", "hive-priv-sock", "libc", - "serde", "serde_json", "tokio", "tracing", diff --git a/flake.nix b/flake.nix index b439cdf3..37f2c26a 100644 --- a/flake.nix +++ b/flake.nix @@ -204,7 +204,6 @@ system treefmt-eval ; - inherit (nixpkgs.lib) nixosSystem; } ); }; diff --git a/hive-agent-mcp/src/mcp/args.rs b/hive-agent-mcp/src/mcp/args.rs index 1f828e76..6a3f5442 100644 --- a/hive-agent-mcp/src/mcp/args.rs +++ b/hive-agent-mcp/src/mcp/args.rs @@ -80,19 +80,6 @@ pub struct RemindArgs { pub file_path: Option, } -/// MCP tool args for `compact`. -#[derive(Debug, serde::Deserialize, schemars::JsonSchema)] -pub struct CompactArgs { - /// Optional wake-up prompt. When set and the compact actually runs - /// (gated on context usage — see the tool description), the harness - /// drives one synthetic follow-up turn with this string as its body - /// as soon as compaction finishes, so you don't have to wait for the - /// next external event to continue. Omit for a fire-and-forget compact - /// with no follow-up. - #[serde(default)] - pub wake_prompt: Option, -} - // ----------------------------------------------------------------------------- // Privileged tool arg types (lifecycle, approvals, scheduling, diagnostics) // ----------------------------------------------------------------------------- diff --git a/hive-agent-mcp/src/mcp/mod.rs b/hive-agent-mcp/src/mcp/mod.rs index 75a86481..4366e59a 100644 --- a/hive-agent-mcp/src/mcp/mod.rs +++ b/hive-agent-mcp/src/mcp/mod.rs @@ -25,10 +25,10 @@ mod render; pub use args::{ AckUntilArgs, AgentGetLooseEndsArgs, AnswerArgs, AskArgs, CancelLooseEndArgs, - CancelScheduleArgs, CompactArgs, CreateRepoArgs, EditScheduleArgs, FireScheduleNowArgs, - GetAgentMetaArgs, GetHostJournalArgs, GetLogsArgs, KillArgs, MarkTodosDoneArgs, RecvArgs, - RemindArgs, RequestInitConfigArgs, RequestSchedulePromptArgs, RestartArgs, SendArgs, - SetStatusArgs, StartArgs, UpdateArgs, UpdateMetaInputsArgs, + CancelScheduleArgs, CreateRepoArgs, EditScheduleArgs, FireScheduleNowArgs, GetAgentMetaArgs, + GetHostJournalArgs, GetLogsArgs, KillArgs, MarkTodosDoneArgs, RecvArgs, RemindArgs, + RequestInitConfigArgs, RequestSchedulePromptArgs, RestartArgs, SendArgs, SetStatusArgs, + StartArgs, UpdateArgs, UpdateMetaInputsArgs, }; pub use render::{annotate_retries, format_ack, format_agent_meta, format_recv}; @@ -657,19 +657,11 @@ impl AgentServer { that the call is refused with an explanation and has no effect. On a pass, \ queues compaction for the end of the current turn (same deferred mechanism \ the dashboard button uses, so it never races a live claude process); the \ - usual pre-compaction notes-checkpoint turn still fires first. Pass \ - `wake_prompt` to have the harness drive one synthetic follow-up turn with \ - that body as soon as compaction finishes — without it you just go idle \ - waiting for the next external event, same as ending a turn normally." + usual pre-compaction notes-checkpoint turn still fires first. No args." )] - async fn compact(&self, Parameters(args): Parameters) -> String { - let log = format!("{args:?}"); - run_tool_envelope("compact", log, async move { - match dial_agent_socket(&hive_agent_sock::Request::Compact { - wake_prompt: args.wake_prompt, - }) - .await - { + async fn compact(&self) -> String { + run_tool_envelope("compact", String::new(), async move { + match dial_agent_socket(&hive_agent_sock::Request::Compact).await { Some(hive_agent_sock::Response::Ok) => { "compact queued — will run at the end of the current turn".to_owned() } diff --git a/hive-agent-sock/src/lib.rs b/hive-agent-sock/src/lib.rs index 8fb6641b..af436056 100644 --- a/hive-agent-sock/src/lib.rs +++ b/hive-agent-sock/src/lib.rs @@ -117,12 +117,7 @@ pub enum Request { /// explains why and takes no action. On a pass, queues the same /// deferred `compact_pending` flag the operator's button sets (consumed /// at the next turn boundary), so it never races a live claude process. - /// `wake_prompt`, when set, is driven as a synthetic follow-up turn once - /// the compaction actually finishes — the dashboard button's own - /// requests go through `Bus::request_compact` directly with `None`, not - /// through this variant, since a human watching the dashboard isn't - /// waiting on a wake. - Compact { wake_prompt: Option }, + Compact, /// Mirror an outstanding question this agent asked (`ask()` succeeded). /// `target` is who it's waiting on (`"operator"` when asked with /// `to: None`). Part of the questions-mirror increment — see diff --git a/hive-agent/src/events.rs b/hive-agent/src/events.rs index e9a26990..cf585f5f 100644 --- a/hive-agent/src/events.rs +++ b/hive-agent/src/events.rs @@ -260,14 +260,6 @@ pub enum TurnState { Compacting, } -/// One pending `/compact` request (see `Bus::request_compact`). -/// `wake_prompt` is what to drive as a synthetic follow-up turn once the -/// compaction actually finishes, if anything. -#[derive(Debug, Clone)] -pub struct CompactRequest { - pub wake_prompt: Option, -} - #[derive(Clone)] pub struct Bus { tx: Arc>, @@ -315,23 +307,7 @@ pub struct Bus { /// One-shot: run `/compact` after the next turn ends. Consumed at the end /// of the current/next turn by `turn::drive_turn`. Deferring to the turn /// boundary keeps compaction from racing a live claude process mid-turn. - /// `Some(request)` when a compact is pending; `request.wake_prompt` is - /// what to drive as a synthetic follow-up turn once the compaction - /// actually completes (`None` = pending but no follow-up wake wanted, - /// e.g. the operator dashboard's `/compact` button). `None` = no compact - /// pending. Wrapped in [`CompactRequest`] rather than - /// `Option>` (clippy pedantic's `option_option` lint, - /// and the named field reads clearer at call sites than a bare nested - /// `Option`) so "pending" and "what to wake with" can never desync. - compact_pending: Arc>>, - /// One-shot, written by `turn::drive_turn`/`turn::run_pending_compact` - /// right after a compaction they served finishes, when that compact's - /// request carried a `wake_prompt`. Read once by the `hive-agent` serve - /// loop after either call site to decide whether to drive a synthetic - /// follow-up turn. Separate from `compact_pending`: by the time this is - /// set, the compact has already run and that flag has already been - /// cleared by `take_compact`. - post_compact_wake: Arc>>, + compact_pending: Arc, /// Current fresh-claude-session id (FK to `sessions.id`). Set by the /// bin loop after minting a session row on a fresh start; stamped onto /// every `turn_stats` row until the next fresh session. `None` before @@ -421,8 +397,7 @@ impl Bus { last_cost_usage: Arc::new(Mutex::new(None)), rate_limited: Arc::new(AtomicBool::new(was_rate_limited)), session_reset_pending: Arc::new(AtomicBool::new(false)), - compact_pending: Arc::new(Mutex::new(None)), - post_compact_wake: Arc::new(Mutex::new(None)), + compact_pending: Arc::new(AtomicBool::new(false)), session_id: Arc::new(Mutex::new(None)), fresh_session: Arc::new(AtomicBool::new(false)), tool_calls: Arc::new(Mutex::new(std::collections::HashMap::new())), @@ -463,40 +438,16 @@ impl Bus { } /// Request a compaction after the next turn ends (deferred to the turn - /// boundary). Idempotent — a second request before the first is - /// serviced just overwrites `wake_prompt` with the latest ask. `Some - /// (wake_prompt)` schedules a synthetic follow-up turn (driven with - /// `wake_prompt` as its body) once the compaction actually completes; - /// `None` requests a plain compact with no follow-up wake (the operator - /// dashboard's `/compact` button). - pub fn request_compact(&self, wake_prompt: Option) { - *self.compact_pending.lock().unwrap() = Some(CompactRequest { wake_prompt }); + /// boundary). Idempotent. + pub fn request_compact(&self) { + self.compact_pending.store(true, Ordering::SeqCst); } - /// Take + clear the compact one-shot. `Some(request)` means - /// `drive_turn`/`run_pending_compact` should compact now — - /// `request.wake_prompt` is what to pass to `set_post_compact_wake` once - /// that compaction finishes. `None` means no compact is pending. + /// Take + clear the compact one-shot. Returns true iff `drive_turn` should + /// compact at the end of this turn. #[must_use] - pub fn take_compact(&self) -> Option { - self.compact_pending.lock().unwrap().take() - } - - /// Record that a just-finished compaction should drive a synthetic - /// follow-up turn with `prompt` as its body. Called by - /// `turn::drive_turn`/`turn::run_pending_compact` right after the - /// compaction they served (whose `take_compact()` returned a request - /// with `wake_prompt: Some(prompt)`) completes. - pub fn set_post_compact_wake(&self, prompt: String) { - *self.post_compact_wake.lock().unwrap() = Some(prompt); - } - - /// Take + clear the post-compact wake one-shot. The serve loop calls - /// this after either compact call site to decide whether to - /// synthesize a follow-up turn. - #[must_use] - pub fn take_post_compact_wake(&self) -> Option { - self.post_compact_wake.lock().unwrap().take() + pub fn take_compact(&self) -> bool { + self.compact_pending.swap(false, Ordering::SeqCst) } /// Mark that the current turn started a fresh claude session. diff --git a/hive-agent/src/main.rs b/hive-agent/src/main.rs index 95106ff4..ce888170 100644 --- a/hive-agent/src/main.rs +++ b/hive-agent/src/main.rs @@ -240,21 +240,6 @@ fn synthetic_todo_message(stern: bool) -> hive_sh4re::inbox::DeliveredMessage { } } -/// Synthetic message that drives the follow-up turn when a `/compact` call -/// (self-requested via the `compact` MCP tool's `wake_prompt` arg) finishes -/// and asked to be woken. Mirrors `synthetic_todo_message`'s "no broker row" -/// sentinel shape (`id = 0`) — the compact tool call itself is the durable -/// record that a wake was requested, not a broker message. -fn post_compact_wake_message(prompt: String) -> hive_sh4re::inbox::DeliveredMessage { - hive_sh4re::inbox::DeliveredMessage { - from: "compact".into(), - body: prompt, - id: 0, - redelivered: false, - in_reply_to: None, - } -} - /// 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. @@ -787,18 +772,8 @@ async fn serve_loop( let compacted = turn::run_pending_compact(files, &bus, &session).await; if !compacted { tokio::time::sleep(interval).await; - continue; } - // The compact that just ran may have carried a wake prompt - // (the agent's own `compact` tool, not the operator button — - // see `Bus::request_compact`). If so, drive it as a synthetic - // turn right now instead of looping back to `recv_next` and - // waiting for the next external event. - let Some(prompt) = bus.take_post_compact_wake() else { - continue; - }; - tracing::debug!("post-compact wake queued, driving synthetic follow-up turn"); - post_compact_wake_message(prompt) + continue; } RecvOutcome::TransportError => { // `recv_next` already logged the detail; just retry. @@ -827,83 +802,32 @@ async fn serve_loop( return Ok(()); } }; - let turn_ctx = TurnCtx { - socket, - bus: &bus, - stats: stats.as_ref(), - files, - session: &session, - interrupted: &interrupted, - login_state: &login_state, - claude_dir: &claude_dir, - interval, - }; - drive_turn_and_wake_chain::(&turn_ctx, next, &mut todo_miss_streak).await; - } -} - -/// Loop-invariant turn-driving context for `drive_turn_and_wake_chain`, -/// threaded as one bundle instead of double-digit positional args -/// (clippy's `too_many_arguments`). Everything here is constant for the -/// lifetime of one `serve_loop` call; only the message to drive and the -/// todo-miss streak vary per turn and stay as separate params. -struct TurnCtx<'a> { - socket: &'a Path, - bus: &'a Bus, - stats: Option<&'a TurnStats>, - files: &'a turn::TurnFiles, - session: &'a turn::AgentSession, - interrupted: &'a Arc, - login_state: &'a Arc>, - claude_dir: &'a Path, - interval: Duration, -} - -/// Drive `next`, then keep driving synthetic follow-up turns for as long as -/// a compact that just ran carries a wake prompt -/// (`Bus::take_post_compact_wake`) — see `serve_loop`'s comment at the call -/// site for why this loops in place instead of returning to the outer -/// `select!`/`recv_next`. Ordinarily runs exactly one iteration; only chains -/// further if the follow-up turn itself requests another woken compact. -/// Split out of `serve_loop` purely to keep that function under clippy's -/// line limit. -async fn drive_turn_and_wake_chain( - ctx: &TurnCtx<'_>, - mut next: hive_sh4re::inbox::DeliveredMessage, - todo_miss_streak: &mut u32, -) { - loop { let ctrl = handle_turn::( - ctx.socket, - ctx.bus, - ctx.stats, - ctx.files, - ctx.session, + socket, + &bus, + stats.as_ref(), + files, + &session, next, - ctx.interrupted, + &interrupted, ) .await; - apply_todo_wake_checked(ctrl.todo_wake_checked, todo_miss_streak, ctx.bus); + apply_todo_wake_checked(ctrl.todo_wake_checked, &mut todo_miss_streak, &bus); if ctrl.auth_failed { - *ctx.login_state.lock().unwrap() = LoginState::NeedsLogin; + *login_state.lock().unwrap() = LoginState::NeedsLogin; // Baseline the resume check on *this instant*, not on a // directory snapshot taken after `wait_for_login` starts // polling — closes the race where a login lands between the // 401 and the first poll. See `wait_for_login`'s doc comment. login::wait_for_login( - ctx.claude_dir, - ctx.login_state.clone(), - ctx.bus, - u64::try_from(ctx.interval.as_millis()).unwrap_or(2000), + &claude_dir, + login_state.clone(), + &bus, + u64::try_from(interval.as_millis()).unwrap_or(2000), std::time::SystemTime::now(), ) .await; } - let Some(prompt) = ctx.bus.take_post_compact_wake() else { - break; - }; - tracing::debug!("post-compact wake queued mid-turn-flow, driving synthetic follow-up turn"); - next = post_compact_wake_message(prompt); } } diff --git a/hive-agent/src/todo_server.rs b/hive-agent/src/todo_server.rs index 307d874d..0352787c 100644 --- a/hive-agent/src/todo_server.rs +++ b/hive-agent/src/todo_server.rs @@ -242,7 +242,7 @@ fn dispatch( } => record_answering_question(questions, id, &asker, &question), Request::ClearQuestion { id } => clear_question(questions, id), Request::ListQuestions => list_questions(questions), - Request::Compact { wake_prompt } => compact(bus, wake_prompt), + Request::Compact => compact(bus), } } @@ -458,10 +458,8 @@ fn mark_todos_done(store: &Todos, ids: &[i64]) -> Response { /// the agent's own MCP tool instead of the dashboard, and refuses below /// [`COMPACT_MIN_USAGE_FRACTION`] instead of always honouring the request — /// an agent can call this speculatively, a human clicking the dashboard -/// button already made the judgment call. `wake_prompt`, when set, is -/// forwarded to `Bus::request_compact` so the turn loop drives a synthetic -/// follow-up turn once the compaction actually finishes. -fn compact(bus: &Bus, wake_prompt: Option) -> Response { +/// button already made the judgment call. +fn compact(bus: &Bus) -> Response { let Some(usage) = bus.last_ctx_usage() else { return Response::Err { message: "compact refused: no completed turn yet — nothing to compact".to_owned(), @@ -489,16 +487,9 @@ fn compact(bus: &Bus, wake_prompt: Option) -> Response { ), }; } - let will_wake = wake_prompt.is_some(); - bus.request_compact(wake_prompt); + bus.request_compact(); bus.emit(crate::events::LiveEvent::Note { - text: if will_wake { - "agent: self-requested /compact (with wake prompt) — running at the end of the \ - current turn" - .into() - } else { - "agent: self-requested /compact — running at the end of the current turn".into() - }, + text: "agent: self-requested /compact — running at the end of the current turn".into(), }); Response::Ok } diff --git a/hive-agent/src/turn.rs b/hive-agent/src/turn.rs index 8b8f4a23..226d1a13 100644 --- a/hive-agent/src/turn.rs +++ b/hive-agent/src/turn.rs @@ -374,18 +374,15 @@ pub async fn drive_turn( archive_session(bus); return Err(TurnError::PromptTooLong); } - // Operator `/compact` (`POST /api/compact`) or an agent's own `compact` - // MCP tool call, deferred to the turn boundary: run it now that the turn - // is done, so it works mid-turn rather than only when the agent is idle. - // Only on a healthy turn — no point spawning a compaction after a - // rate-limited / auth-failed / crashed one. `is_ok()` first: `take_compact()` - // clears the flag, so it must only fire when the compaction will actually - // run. On an unhealthy turn (rate-limited / auth-failed / failed) the flag - // is left set for the next turn or the idle `run_pending_compact` to - // service — not silently eaten. - if outcome.is_ok() - && let Some(request) = bus.take_compact() - { + // Operator `/compact` (`POST /api/compact`) deferred to the turn boundary: + // run it now that the turn is done, so it works mid-turn rather than only + // when the agent is idle. Only on a healthy turn — no point spawning a + // compaction after a rate-limited / auth-failed / crashed one. + // `is_ok()` first: `take_compact()` clears the flag, so it must only fire + // when the compaction will actually run. On an unhealthy turn + // (rate-limited / auth-failed / failed) the flag is left set for the next + // turn or the idle `run_pending_compact` to service — not silently eaten. + if outcome.is_ok() && bus.take_compact() { bus.emit(LiveEvent::Note { text: "operator: /compact — running at turn end".into(), }); @@ -393,14 +390,6 @@ pub async fn drive_turn( // does; the serve loop resets to `Idle` once this turn returns. bus.set_state(crate::events::TurnState::Compacting); let _ = session.compact(&config, &sink).await; - // If the compact call asked to be woken (the agent's own `compact` - // tool with a `wake_prompt`), stash it — the serve loop reads it - // back after this turn returns and drives a synthetic follow-up - // turn, so a self-requested compact provably doesn't strand the - // agent idle waiting for the next external event. - if let Some(prompt) = request.wake_prompt { - bus.set_post_compact_wake(prompt); - } return Ok(true); } outcome @@ -511,13 +500,11 @@ pub fn emit_turn_end(bus: &Bus, outcome: &TurnOutcome) { /// so a queued `/compact` runs even when no turn is driving. (The in-flight /// case is handled at the end of [`drive_turn`].) Resume-only via /// [`InfiniteSession::compact`]: a missing session is a harmless no-op. Returns -/// `true` if a compaction ran; the serve loop follows up with -/// `Bus::take_post_compact_wake` to see whether a synthetic follow-up turn -/// should run (set below when the compact request carried a `wake_prompt`). +/// `true` if a compaction ran. pub async fn run_pending_compact(files: &TurnFiles, bus: &Bus, session: &AgentSession) -> bool { - let Some(request) = bus.take_compact() else { + if !bus.take_compact() { return false; - }; + } bus.emit(LiveEvent::Note { text: "operator: /compact — running on idle session".into(), }); @@ -533,9 +520,6 @@ pub async fn run_pending_compact(files: &TurnFiles, bus: &Bus, session: &AgentSe }), } bus.set_state(crate::events::TurnState::Idle); - if let Some(prompt) = request.wake_prompt { - bus.set_post_compact_wake(prompt); - } true } diff --git a/hive-agent/src/web_ui/actions.rs b/hive-agent/src/web_ui/actions.rs index c5e7d115..56c7ea8d 100644 --- a/hive-agent/src/web_ui/actions.rs +++ b/hive-agent/src/web_ui/actions.rs @@ -7,7 +7,7 @@ use axum::{ http::StatusCode, response::{IntoResponse, Response}, }; -use serde::{Deserialize, Serialize}; +use serde::Deserialize; use super::{AppState, SigintOutcome, error_response}; @@ -80,10 +80,7 @@ pub(super) async fn post_cancel_turn(State(state): State) -> Response /// claude process) rather than only when the agent is idle. Returns 200 /// immediately; the compaction stream lands in the live panel when it runs. pub(super) async fn post_compact(State(state): State) -> Response { - // No wake prompt: the operator is watching the dashboard, not waiting on - // an inbox message — `request_compact`'s wake-prompt arg exists for the - // agent's own `compact` MCP tool (`todo_server.rs::compact`). - state.bus.request_compact(None); + state.bus.request_compact(); state.bus.emit(crate::events::LiveEvent::Note { text: "operator: /compact queued — runs at the end of the current turn".into(), }); @@ -205,10 +202,5 @@ pub(super) async fn post_mark_todos_done(Form(form): Form) -> acked += count; } } - axum::Json(MarkTodosDoneBody { acked }).into_response() -} - -#[derive(Serialize)] -struct MarkTodosDoneBody { - acked: u64, + axum::Json(serde_json::json!({ "acked": acked })).into_response() } diff --git a/hive-agent/src/web_ui/stats.rs b/hive-agent/src/web_ui/stats.rs index 9dd6796d..f2f83614 100644 --- a/hive-agent/src/web_ui/stats.rs +++ b/hive-agent/src/web_ui/stats.rs @@ -2,7 +2,7 @@ use axum::extract::State; use axum::response::{IntoResponse, Response}; -use serde::{Deserialize, Serialize}; +use serde::Deserialize; use super::AppState; @@ -40,11 +40,6 @@ async fn fetch_reminder_stats(window_secs: u64) -> Option, -} - /// `GET /api/todos` — snapshot of this agent's local todos (loose-ends v2). /// /// Connects to the in-agent harness socket (`HIVE_AGENT_SOCKET`) and calls @@ -59,5 +54,5 @@ pub(super) async fn api_todos() -> Response { Some(hive_agent_sock::Response::LooseEnds { loose_ends }) => loose_ends, _ => Vec::new(), }; - axum::Json(TodosBody { todos }).into_response() + axum::Json(serde_json::json!({ "todos": todos })).into_response() } diff --git a/hive-agent/src/web_ui/stream.rs b/hive-agent/src/web_ui/stream.rs index 89c4ba31..3c881cd4 100644 --- a/hive-agent/src/web_ui/stream.rs +++ b/hive-agent/src/web_ui/stream.rs @@ -5,23 +5,11 @@ use std::convert::Infallible; use axum::Json; use axum::extract::{Query, State}; use axum::response::sse::{Event, KeepAlive, Sse}; -use serde::{Deserialize, Serialize}; +use serde::Deserialize; use tokio_stream::{Stream, StreamExt, wrappers::BroadcastStream}; use super::AppState; -/// Response body for `GET /api/events/history`. `seq` is omitted from the -/// wire entirely on a paginated (non-initial) load — matches the old -/// `json!` shape, which only ever set the `"seq"` key when `Some`. -#[derive(Serialize)] -pub(super) struct EventsHistoryBody { - events: Vec, - min_id: Option, - has_more: bool, - #[serde(skip_serializing_if = "Option::is_none")] - seq: Option, -} - /// Query params for the paginated history endpoint. #[derive(Debug, Deserialize)] pub(super) struct HistoryParams { @@ -35,7 +23,7 @@ pub(super) struct HistoryParams { pub(super) async fn events_history( State(state): State, Query(params): Query, -) -> Json { +) -> Json { use crate::events::HISTORY_CAPACITY; let limit = params.limit.unwrap_or(100).min(HISTORY_CAPACITY); let before = params.before; @@ -63,12 +51,15 @@ pub(super) async fn events_history( se }) .collect(); - Json(EventsHistoryBody { - events, - min_id, - has_more, - seq, - }) + let mut resp = serde_json::json!({ + "events": events, + "min_id": min_id, + "has_more": has_more, + }); + if let Some(s) = seq { + resp["seq"] = serde_json::json!(s); + } + Json(resp) } pub(super) async fn events_stream( diff --git a/hive-c0re/src/actions.rs b/hive-c0re/src/actions.rs index cdf84e88..085d2bf7 100644 --- a/hive-c0re/src/actions.rs +++ b/hive-c0re/src/actions.rs @@ -899,20 +899,99 @@ pub async fn run_finalize_deploy(coord: &Arc, approval_id: i64) -> /// imperative infra that `auto_update::ensure_root_agent` recreates on the /// next hive-c0re startup if absent, so destroying it is transient rather /// than something to refuse at the API. -/// -/// Submits the teardown DAG and returns — it does not wait for the container to -/// go away. Same contract as every other lifecycle op (`rebuild`, `kill`, -/// `restart`, `start`): the queue owns the work, the caller gets an -/// acknowledgement. Progress is visible as real nodes on the dashboard. -pub fn destroy(coord: &Arc, name: &str, purge: bool) { +pub async fn destroy(coord: &Arc, name: &str, purge: bool) -> Result<()> { tracing::info!(%name, purge, "destroy"); - if let Err(e) = coord.job_queue.insert_job(|b| { - crate::job_queue::templates::destroy(b, name, purge); - Vec::new() - }) { - tracing::error!(agent = %name, error = ?e, "destroy: insert failed"); + // Guard auto-clears on the success path's final scope exit and on + // every early-return / cancellation along the way. + // Destroy has no queue node behind it, so nothing in the graph says this + // container is going away on purpose — without this the crash watcher + // reports every destroy as a crash and the manager tries to recover it. + let guard = coord.suppress_crash_watch(name); + lifecycle::destroy(name).await?; + coord.unregister_agent(name); + let runtime = crate::paths::agent_runtime_dir(name); + if runtime.exists() { + let _ = std::fs::remove_dir_all(&runtime); } - coord.emit_rebuild_queue_snapshot(); + if purge { + // The state root may be a btrfs subvolume: a subvolume root + // can't be removed with rmdir/`remove_dir_all`, so delete it via + // hive-priv (root) first. No-op for plain-dir agents — the loop below + // then handles the plain-dir state root plus the applied dir. + if let Err(e) = crate::priv_client::delete_agent_subvolume(name).await { + tracing::warn!(error = ?e, %name, "purge: delete state subvolume failed"); + } + // A malformed name can't have a persistent state tree (the state dir + // is only ever created under a validated Ident), so its removal is a + // no-op — skip the state-dir sweep and just clear the applied dir. + let state_dir = hive_types::Ident::parse(name) + .ok() + .map(|id| crate::paths::agent_state_dir(&id)); + for dir in state_dir + .into_iter() + .chain([crate::paths::applied_dir(name)]) + { + if dir.exists() + && let Err(e) = std::fs::remove_dir_all(&dir) + { + tracing::warn!(error = ?e, dir = %dir.display(), "purge: remove failed"); + } + } + } + // Meta flake: drop the agent's input + nixosConfiguration so a + // future spawn under the same name re-seeds cleanly, and so the + // meta lock doesn't reference a vanished applied repo. Log + keep + // going on failure — destroy already succeeded at the + // nixos-container level, the meta repo is just bookkeeping. + if let Err(e) = sync_meta_after_lifecycle(coord).await { + tracing::warn!(error = ?e, %name, "meta sync after destroy failed"); + } + let _ = coord.approvals.fail_pending_for_agent( + name, + if purge { + "agent purged" + } else { + "agent destroyed" + }, + ); + // Drop the durable power intent — a future agent of the same name + // seeds fresh from its observed state. + if let Err(e) = coord.power.remove(name) { + tracing::warn!(%name, error = ?e, "agent_power: remove on destroy failed"); + } + drop(guard); + let _ = coord + .push_todo( + hive_sh4re::manager::MANAGER_AGENT, + "core", + Some(format!("destroyed:{name}")), + format!("agent '{name}' destroyed"), + None, + false, + ) + .await; + // Container row disappeared — rescan so the dashboard fires + // `ContainerRemoved` for the gone row, then emit the + // tombstones snapshot (gained one on destroy, lost one on + // purge — recompute either way). + coord.rescan_containers_and_emit().await; + crate::dashboard::emit_tombstones_snapshot(coord).await; + // Re-emit the schedules snapshot: the rescan above refreshed the live + // roster, so any schedule that still targets the just-destroyed agent + // now drops that ghost column live (no page reload needed). + coord.emit_schedules_snapshot(); + // Update tmpfiles.d to remove the destroyed agent's dirs from the + // boot-time pre-creation list. Best-effort: failure is logged only. + tokio::spawn(lifecycle::sync_tmpfiles()); + Ok(()) +} + +/// Rerender the meta flake from whatever containers still exist on +/// disk. Called after lifecycle ops that change the agent set (today: +/// destroy). Idempotent — a no-op when nothing changed. +async fn sync_meta_after_lifecycle(coord: &Coordinator) -> Result<()> { + let agents = lifecycle::agents_for_meta_listing().await?; + crate::meta::sync_agents(&coord.hive_env(), &agents).await } pub fn deny(coord: &Coordinator, id: i64, note: Option<&str>) -> Result<()> { diff --git a/hive-c0re/src/coordinator.rs b/hive-c0re/src/coordinator.rs index 4f1cf362..1e8eb46e 100644 --- a/hive-c0re/src/coordinator.rs +++ b/hive-c0re/src/coordinator.rs @@ -116,6 +116,13 @@ pub struct Coordinator { /// is never injected into containers. pub model_prices: crate::hive_stats::PriceTable, agents: Mutex>, + /// Agents whose lifecycle action (currently just spawn) is in flight. + /// Read by the dashboard to render a spinner; cleared when the action + /// resolves (success or failure). + /// Agents whose container is being taken down by work with **no queue node + /// behind it** (destroy, migration), so the crash watcher must not report + /// the disappearance as a crash. Not a pill — see [`CrashWatchSuppression`]. + crash_suppressed: Mutex>, /// Tombstone for transients that have JUST been cleared. The /// crash watcher polls every 10s and would race the /// drop-clears-immediately path of `TransientGuard`: an operator @@ -137,7 +144,9 @@ pub struct Coordinator { /// live and both clear. Keyed by agent, the last clear *overwrites* the /// others: a `Prebuild` (`deliberate_stop = false`) landing after a /// `StopForUpdate` (`true`) leaves the tombstone reading `false`, and the - /// crash watcher then reports an intentional stop as a **crash**. + /// crash watcher then reports an intentional stop as a **crash**. The + /// out-of-band suppression guard, which has no node behind it, uses + /// [`NO_NODE_LABEL`]. recent_transient: Mutex>, /// Timestamps of recent unexpected container crashes, keyed by agent. /// Fed by `crash_watch` each time it classifies a stop as a crash (so @@ -391,6 +400,57 @@ fn fold_tombstones_by_agent<'a>( out } +/// Tombstone label for work with **no queue node behind it** — the +/// out-of-band operations (destroy, migration) that hold a +/// [`CrashWatchGuard`] instead of appearing in the derived transient set. +/// +/// [`Coordinator::recent_transient`] is keyed by `(agent, label)` so concurrent +/// pills can't overwrite each other's `deliberate_stop`; a guard has no node and +/// therefore no node label, so it needs one of its own. Angle-bracketed to keep +/// it out of the `NodeKind::as_str` namespace — no node can ever render this. +const NO_NODE_LABEL: &str = ""; + +/// RAII handle returned by [`Coordinator::suppress_crash_watch`]. While held, +/// the crash watcher treats this container disappearing as **expected**. +/// +/// This is *not* a dashboard pill. Transients are derived from running queue +/// nodes and nothing stores them. But destroy and migration take a container +/// down without a node behind them, so nothing in the graph says the +/// disappearance was intended — and without that, `crash_watch` fires a +/// `ContainerCrash` for every destroy and every migrated agent, and the manager +/// tries to "recover" containers that were removed on purpose. +/// +/// It is held rather than stamped once because +/// [`crate::workers::crash_watch`]'s grace window is finite and these +/// operations are not: a long destroy would outlive a single tombstone. The +/// tombstone is stamped on drop, covering the poll that lands just after. +/// +/// Goes away entirely once destroy + migration are real queue nodes. +#[must_use = "suppression lasts as long as the guard; bind it for the operation's duration \ + (`let _guard = coord.suppress_crash_watch(...)`). An unbound call drops it \ + immediately and the very next poll can report a deliberate stop as a crash."] +pub struct CrashWatchSuppression { + coord: Arc, + name: String, +} + +impl Drop for CrashWatchSuppression { + fn drop(&mut self) { + self.coord + .crash_suppressed + .lock() + .unwrap() + .remove(&self.name); + // Tombstone the release so the next poll — which may land in the + // window between the container going away and this guard dropping — + // still reads the stop as deliberate. + self.coord.recent_transient.lock().unwrap().insert( + (self.name.clone(), NO_NODE_LABEL.to_owned()), + (true, std::time::Instant::now()), + ); + } +} + /// RAII guard for the `meta-update` in-progress flag, held for the /// duration of a `run_meta_update` background task. Created by /// `Coordinator::meta_update_guard`. Drop decrements the active-run @@ -526,6 +586,7 @@ impl Coordinator { agent_io_weight, model_prices, agents: Mutex::new(HashMap::new()), + crash_suppressed: Mutex::new(HashSet::new()), recent_transient: Mutex::new(HashMap::new()), recent_crashes: Mutex::new(HashMap::new()), graceful_stop_pending: Mutex::new(HashSet::new()), @@ -1239,16 +1300,46 @@ impl Coordinator { map.iter().map(|(k, v)| (k.clone(), v.len())).collect() } + /// Tell the crash watcher that `name`'s container is going down **on + /// purpose**, for the lifetime of the returned guard. See + /// [`CrashWatchSuppression`] for why this exists at all. + /// + /// Only for the operations with no queue node behind them. Anything the + /// job queue runs answers this from the node itself + /// ([`crate::job_queue::NodeKind::takes_container_down`]) and must not come + /// through here. + /// + /// The guard's `Drop` runs even on task cancellation, so an aborted HTTP + /// request or a panic mid-destroy can't leave a container permanently + /// exempt from crash reporting. + pub fn suppress_crash_watch(self: &Arc, name: &str) -> CrashWatchSuppression { + self.crash_suppressed + .lock() + .unwrap() + .insert(name.to_owned()); + CrashWatchSuppression { + coord: self.clone(), + name: name.to_owned(), + } + } + + /// Whether a no-node operation is currently taking this container down. + #[must_use] + pub fn crash_watch_suppressed(&self, name: &str) -> bool { + self.crash_suppressed.lock().unwrap().contains(name) + } + /// Every live transient, keyed by agent. /// /// **Derived on read, stored nowhere.** Straight off the running graph, so /// there is no cached copy to go stale, leak, or disagree with what is /// actually running. /// - /// Migration is the last operation with no queue node behind it, so it - /// shows **no pill** — there is nothing in the graph to derive one from. - /// It gets one for free once it becomes real nodes, the way destroy did. - /// + /// Work with no queue node behind it (destroy, migration) therefore shows + /// **no pill** — there is nothing in the graph to derive one from. Its + /// crash-watch suppression is a separate, narrower thing + /// ([`Coordinator::suppress_crash_watch`]); the pill comes back for free + /// once those become real nodes. /// ⚠️ **A `Vec` per agent, not one entry.** `running_transients` tests /// status alone, so a lease-exempt `Prebuild` for `a` and a lease-holding /// `StopForUpdate` for `a` are both live pills. Collapsing them to one @@ -1476,13 +1567,8 @@ impl Coordinator { crate::paths::agent_runtime_dir(name).join("mcp.sock") } - /// The *proposed* config repo: where a config change lands before it is - /// applied, and what an approved deploy promotes into `applied_dir`. - /// - /// **Not bind-mounted into any container.** An agent that edits a config - /// clones it from the forge itself; `/agents//config` shows the - /// applied (deployed) tree instead — see `config_bind_source` in - /// `lifecycle/host_config.rs`. + /// Manager-editable proposed config repo. Bind-mounted into the manager + /// container as `/agents//config/`. pub fn agent_proposed_dir(name: &hive_types::Ident) -> PathBuf { crate::paths::agent_state_dir(name).join("config") } diff --git a/hive-c0re/src/dashboard/health.rs b/hive-c0re/src/dashboard/health.rs index 6f5837fc..f1cc19a8 100644 --- a/hive-c0re/src/dashboard/health.rs +++ b/hive-c0re/src/dashboard/health.rs @@ -28,20 +28,19 @@ use utoipa::ToSchema; use crate::host_stats::ServerWarning; -#[derive(Serialize, ToSchema)] -struct LiveBody { - status: &'static str, -} - /// Liveness. Always `200`; no further checks. #[utoipa::path( get, path = "/health/live", - responses((status = 200, description = "process is up", body = LiveBody)), + responses((status = 200, description = "process is up", body = serde_json::Value)), tag = "health" )] pub(super) async fn get_health_live() -> Response { - (StatusCode::OK, axum::Json(LiveBody { status: "ok" })).into_response() + ( + StatusCode::OK, + axum::Json(serde_json::json!({ "status": "ok" })), + ) + .into_response() } #[derive(Serialize, ToSchema)] diff --git a/hive-c0re/src/dashboard/lifecycle_ops.rs b/hive-c0re/src/dashboard/lifecycle_ops.rs index 855c2830..e12723fa 100644 --- a/hive-c0re/src/dashboard/lifecycle_ops.rs +++ b/hive-c0re/src/dashboard/lifecycle_ops.rs @@ -436,9 +436,10 @@ pub(super) struct DestroyForm { params(("name" = String, Path, description = "agent name")), request_body(content = DestroyForm, content_type = "application/x-www-form-urlencoded"), responses( - (status = 200, description = "destroy queued", body = String), + (status = 200, description = "destroyed", body = String), (status = 400, description = "bad agent name"), (status = 404, description = "no such agent"), + (status = 500, description = "destroy failed"), ), tag = "lifecycle_ops" )] @@ -452,10 +453,11 @@ pub(super) async fn post_destroy( } // Checkbox semantics: any non-empty value (axum sends "on") = purge. let purge = form.purge.as_deref().is_some_and(|v| !v.is_empty()); - // Submit-and-return, like every other lifecycle endpoint here. The - // container rescan now runs in the DAG's bookkeeping tail, so - // `ContainerRemoved` arrives *after* this 200 rather than before it — the - // row disappears when the event lands, same as a rebuild's does. - actions::destroy(&state.coord, &name, purge); - (StatusCode::OK, "ok").into_response() + // `actions::destroy` rescans the container list on success, so the + // `ContainerRemoved` event lands before we return 200. The matching + // form carries `data-no-refresh`. + match actions::destroy(&state.coord, &name, purge).await { + Ok(()) => (StatusCode::OK, "ok").into_response(), + Err(e) => error_response(&format!("destroy {name} failed: {e:#}")), + } } diff --git a/hive-c0re/src/dashboard/misc_api.rs b/hive-c0re/src/dashboard/misc_api.rs index 51695b9a..c335cae4 100644 --- a/hive-c0re/src/dashboard/misc_api.rs +++ b/hive-c0re/src/dashboard/misc_api.rs @@ -8,41 +8,26 @@ use axum::{ http::StatusCode, response::{IntoResponse, Response}, }; -use serde::{Deserialize, Serialize}; +use serde::Deserialize; use utoipa::{IntoParams, ToSchema}; use super::{AppState, Ident, error_response, scan_validated_paths}; -use crate::audit_log::AuditEntry; use crate::container_stats::ContainerResource; use crate::hive_stats::HiveStats; -#[derive(Serialize, ToSchema)] -pub(super) struct OperatorInboxItem { - id: i64, - from: String, - body: String, - at: chrono::DateTime, - in_reply_to: Option, - file_refs: Vec, -} - -#[derive(Serialize, ToSchema)] -pub(super) struct OperatorInboxBody { - messages: Vec, -} - /// Unread operator-directed messages for the dashboard's Y3R C4LL inbox. /// /// Returns messages addressed to `"operator"` that haven't been /// acked yet (the operator clears them via the existing /// `POST /api/agent/operator/mark-all-read`). Newest-first; path-shaped /// tokens are validated so the client renders file links like the -/// terminal does. +/// terminal does. Shape: `{ "messages": [{ id, from, body, at, +/// in_reply_to, file_refs }] }`. #[utoipa::path( get, path = "/api/operator-inbox", responses( - (status = 200, description = "unread operator-directed messages", body = OperatorInboxBody), + (status = 200, description = "unread operator-directed messages", body = serde_json::Value), (status = 500, description = "broker read failed"), ), tag = "misc_api" @@ -55,7 +40,7 @@ pub(super) async fn api_operator_inbox(State(state): State) -> Respons .unread_for_recipient("operator", INBOX_LIMIT) { Ok(messages) => { - let messages: Vec = messages + let items: Vec = messages .into_iter() .filter_map(|m| { let crate::broker::MessageEvent::Sent { @@ -70,17 +55,17 @@ pub(super) async fn api_operator_inbox(State(state): State) -> Respons return None; }; let file_refs = scan_validated_paths(&body); - Some(OperatorInboxItem { - id, - from, - at: hive_sh4re::wire_time::from_secs(at), - body, - in_reply_to, - file_refs, - }) + Some(serde_json::json!({ + "id": id, + "from": from, + "body": body, + "at": hive_sh4re::wire_time::from_secs(at), + "in_reply_to": in_reply_to, + "file_refs": file_refs, + })) }) .collect(); - axum::Json(OperatorInboxBody { messages }).into_response() + axum::Json(serde_json::json!({ "messages": items })).into_response() } Err(e) => error_response(&format!("operator-inbox failed: {e:#}")), } @@ -129,23 +114,18 @@ pub(super) async fn api_container_resources() -> Response { axum::Json(crate::container_stats::gather().await).into_response() } -#[derive(Serialize, ToSchema)] -pub(super) struct AuditLogBody { - entries: Vec, - total: i64, -} - /// Most-recent agent-initiated privileged-action /// audit entries, newest first (server-clamped to 500). /// -/// Backs the operator dashboard's audit view. `total` lets the UI show +/// Backs the operator dashboard's audit view. Returns +/// `{ "entries": [AuditEntry…], "total": N }` so the UI can show /// "latest 500 of N" rather than silently capping. `ts_unix` is in /// **seconds**. #[utoipa::path( get, path = "/api/audit-log", responses( - (status = 200, description = "recent audit entries + total count", body = AuditLogBody), + (status = 200, description = "recent audit entries + total count", body = serde_json::Value), (status = 500, description = "sqlite read failed"), ), tag = "misc_api" @@ -160,12 +140,7 @@ pub(super) async fn api_audit_log(State(state): State) -> Response { Ok(n) => n, Err(e) => return error_response(&format!("audit-log count: {e:#}")), }; - axum::Json(AuditLogBody { entries, total }).into_response() -} - -#[derive(Serialize, ToSchema)] -pub(super) struct MarkAllReadBody { - marked: u64, + axum::Json(serde_json::json!({ "entries": entries, "total": total })).into_response() } /// Operator-driven "clear this agent's inbox" — backs the side-panel @@ -173,14 +148,14 @@ pub(super) struct MarkAllReadBody { /// /// Marks every message addressed to the agent as acked (backfilling /// `delivered_at` for any still-pending rows so vacuum can collect -/// them). `marked` lets the frontend show "cleared N messages" -/// feedback without an extra fetch. +/// them). Returns `{ "marked": N }` so the frontend can show "cleared +/// N messages" feedback without an extra fetch. #[utoipa::path( post, path = "/api/agent/{name}/mark-all-read", params(("name" = String, Path, description = "agent name")), responses( - (status = 200, description = "count of messages marked read", body = MarkAllReadBody), + (status = 200, description = "count of messages marked read", body = serde_json::Value), (status = 400, description = "bad agent name"), (status = 500, description = "broker write failed"), ), @@ -197,9 +172,9 @@ pub(super) async fn post_mark_all_read( } }; match state.coord.broker.mark_all_read(name.as_str()) { - Ok(marked) => { - tracing::info!(%name, marked, "operator marked all messages read"); - axum::Json(MarkAllReadBody { marked }).into_response() + Ok(n) => { + tracing::info!(%name, marked = n, "operator marked all messages read"); + axum::Json(serde_json::json!({ "marked": n })).into_response() } Err(e) => error_response(&format!("mark-all-read {name} failed: {e:#}")), } diff --git a/hive-c0re/src/dashboard/schedules.rs b/hive-c0re/src/dashboard/schedules.rs index 79322a72..6ea9ea30 100644 --- a/hive-c0re/src/dashboard/schedules.rs +++ b/hive-c0re/src/dashboard/schedules.rs @@ -18,16 +18,6 @@ use crate::scheduled_prompts_worker::FireNowReport; use super::{AppState, error_problem, error_response}; -#[derive(serde::Serialize, utoipa::ToSchema)] -pub(super) struct NewScheduleBody { - id: i64, -} - -#[derive(serde::Serialize, utoipa::ToSchema)] -pub(super) struct CancelResultBody { - cancelled: bool, -} - /// Snapshot of every schedule for the /// scheduled-prompts tab. /// @@ -84,7 +74,7 @@ pub(super) async fn api_schedules(State(state): State) -> Response { // `api_schedules` above. request_body(content = serde_json::Value, description = "SchedulePromptPayload wire shape"), responses( - (status = 200, description = "created; body carries the new row id", body = NewScheduleBody), + (status = 200, description = "created; body carries the new row id", body = serde_json::Value), (status = 400, description = "no targets, empty body, or interval_seconds == 0"), (status = 500, description = "submit failed"), ), @@ -118,7 +108,7 @@ pub(super) async fn post_schedule_new( match state.coord.scheduled_prompts.submit(&new) { Ok(id) => { state.coord.emit_schedules_snapshot(); - Ok(axum::Json(NewScheduleBody { id }).into_response()) + Ok(axum::Json(serde_json::json!({"id": id})).into_response()) } Err(e) => Err(error_problem(&format!("schedule submit: {e:#}"))), } @@ -187,7 +177,7 @@ pub(super) async fn post_schedule_fire_now( post, path = "/api/rebuild-queue/{id}/cancel", params(("id" = u64, Path, description = "job-queue node id (a DAG's root cancels the group)")), - responses((status = 200, description = "whether the DAG was cancelled", body = CancelResultBody)), + responses((status = 200, description = "whether the DAG was cancelled", body = serde_json::Value)), tag = "schedules" )] pub(super) async fn post_rebuild_queue_cancel( @@ -198,9 +188,9 @@ pub(super) async fn post_rebuild_queue_cancel( // Any terminal side effect is the DAG's own spared tail node, which the // scheduler picks up on its next pass — nothing to fire from here. state.coord.emit_rebuild_queue_snapshot(); - axum::Json(CancelResultBody { cancelled: true }).into_response() + axum::Json(serde_json::json!({"cancelled": true})).into_response() } else { - axum::Json(CancelResultBody { cancelled: false }).into_response() + axum::Json(serde_json::json!({"cancelled": false})).into_response() } } diff --git a/hive-c0re/src/dashboard/state_snapshot.rs b/hive-c0re/src/dashboard/state_snapshot.rs index 56b74cc5..d3d4bc08 100644 --- a/hive-c0re/src/dashboard/state_snapshot.rs +++ b/hive-c0re/src/dashboard/state_snapshot.rs @@ -726,18 +726,6 @@ pub(super) async fn jobq_rollup( axum::Json(state.coord.job_queue.state_rollup()) } -/// Response body for `/api/dashboard/history`. No `ToSchema` — its -/// `events` field wraps [`crate::dashboard_events::DashboardEvent`], -/// which doesn't derive `ToSchema` either (a large enum with many -/// variants; see that type's doc comment for why annotating it is -/// out of scope here). The `responses(...)` doc below spells out the -/// shape in prose instead of a `body = ...` reference. -#[derive(Serialize)] -struct DashboardHistoryBody { - seq: u64, - events: Vec, -} - #[utoipa::path( get, path = "/api/dashboard/history", @@ -810,7 +798,7 @@ pub(super) async fn dashboard_history(State(state): State) -> Response } }) .collect(); - axum::Json(DashboardHistoryBody { seq, events }).into_response() + axum::Json(serde_json::json!({ "seq": seq, "events": events })).into_response() } Err(e) => error_response(&format!("dashboard/history failed: {e:#}")), } diff --git a/hive-c0re/src/job_queue/exec.rs b/hive-c0re/src/job_queue/exec.rs index 9efcf81c..d392bd78 100644 --- a/hive-c0re/src/job_queue/exec.rs +++ b/hive-c0re/src/job_queue/exec.rs @@ -73,15 +73,6 @@ pub(super) async fn run_node( NodeKind::RebuildBookkeeping { .. } => run_rebuild_bookkeeping(coord, agent).await, NodeKind::Provision { .. } => run_provision(coord, agent).await, NodeKind::Create { .. } => run_create(agent).await, - NodeKind::DestroyContainer { .. } => run_destroy_container(coord, agent).await, - NodeKind::PurgeState { .. } => { - run_purge_state(agent).await; - Ok(()) - } - NodeKind::DestroyBookkeeping { purge, .. } => { - run_destroy_bookkeeping(coord, agent, *purge).await; - Ok(()) - } NodeKind::MetaLock { sweep, fanout, @@ -172,112 +163,6 @@ async fn run_resolve_approval( Ok(()) } -/// Rerender the meta flake from whatever containers still exist on disk. -/// Idempotent — a no-op when nothing changed. Lives here because the destroy -/// tail is its only caller; it moved with `destroy` when that became a DAG. -async fn sync_meta_after_lifecycle(coord: &Coordinator) -> Result<()> { - let agents = crate::lifecycle::agents_for_meta_listing().await?; - crate::meta::sync_agents(&coord.hive_env(), &agents).await -} - -/// `nixos-container destroy`, then drop the agent from the roster and clear -/// its ephemeral runtime dir (the mcp socket, which does not survive a restart -/// anyway). -/// -/// The only fallible step is the destroy itself: once the container is gone the -/// un-registration cannot meaningfully fail, and returning early would strand -/// the roster claiming an agent that no longer exists. -async fn run_destroy_container(coord: &Arc, agent: &str) -> Result<()> { - crate::lifecycle::destroy(agent).await?; - coord.unregister_agent(agent); - let runtime = crate::paths::agent_runtime_dir(agent); - if runtime.exists() { - let _ = std::fs::remove_dir_all(&runtime); - } - Ok(()) -} - -/// The `purge = true` half: wipe the agent's persistent trees. -/// -/// Every step is best-effort-with-a-warning rather than fatal, and that is -/// deliberate — the container is already destroyed by the time this runs, so -/// failing the node would leave the operator with a half-purged agent and a red -/// DAG, when what they can actually act on is the log line naming the path. -async fn run_purge_state(agent: &str) { - // The state root may be a btrfs subvolume: a subvolume root can't be - // removed with rmdir/`remove_dir_all`, so delete it via hive-priv (root) - // first. No-op for plain-dir agents — the loop below then handles the - // plain-dir state root plus the applied dir. - if let Err(e) = crate::priv_client::delete_agent_subvolume(agent).await { - tracing::warn!(error = ?e, %agent, "purge: delete state subvolume failed"); - } - // A malformed name can't have a persistent state tree (the state dir is - // only ever created under a validated Ident), so its removal is a no-op — - // skip the state-dir sweep and just clear the applied dir. - let state_dir = hive_types::Ident::parse(agent) - .ok() - .map(|id| crate::paths::agent_state_dir(&id)); - for dir in state_dir - .into_iter() - .chain([crate::paths::applied_dir(agent)]) - { - if dir.exists() - && let Err(e) = std::fs::remove_dir_all(&dir) - { - tracing::warn!(error = ?e, dir = %dir.display(), "purge: remove failed"); - } - } -} - -/// Post-destroy bookkeeping. Infallible by construction: every step is -/// warn-and-continue, because the destroy it follows has already succeeded and -/// none of this is undoable — a failed meta sync or power-store write is a -/// bookkeeping drift to log, not a reason to red a DAG whose container is -/// already gone. -async fn run_destroy_bookkeeping(coord: &Arc, agent: &str, purge: bool) { - // Meta flake: drop the agent's input + nixosConfiguration so a future spawn - // under the same name re-seeds cleanly, and so the meta lock doesn't - // reference a vanished applied repo. - if let Err(e) = sync_meta_after_lifecycle(coord).await { - tracing::warn!(error = ?e, %agent, "meta sync after destroy failed"); - } - let _ = coord.approvals.fail_pending_for_agent( - agent, - if purge { - "agent purged" - } else { - "agent destroyed" - }, - ); - // Drop the durable power intent — a future agent of the same name seeds - // fresh from its observed state. - if let Err(e) = coord.power.remove(agent) { - tracing::warn!(%agent, error = ?e, "agent_power: remove on destroy failed"); - } - let _ = coord - .push_todo( - hive_sh4re::manager::MANAGER_AGENT, - "core", - Some(format!("destroyed:{agent}")), - format!("agent '{agent}' destroyed"), - None, - false, - ) - .await; - // Container row disappeared — rescan so the dashboard fires - // `ContainerRemoved` for the gone row, then emit the tombstones snapshot - // (gained one on destroy, lost one on purge — recompute either way). - coord.rescan_containers_and_emit().await; - crate::dashboard::emit_tombstones_snapshot(coord).await; - // Re-emit the schedules snapshot: the rescan above refreshed the live - // roster, so any schedule that still targets the just-destroyed agent now - // drops that ghost column live (no page reload needed). - coord.emit_schedules_snapshot(); - // Update tmpfiles.d to remove the destroyed agent's dirs from the boot-time - // pre-creation list. Best-effort: failure is logged only. - tokio::spawn(crate::lifecycle::sync_tmpfiles()); -} - /// Emit this agent's rebuild-complete todo. `ok` is not computed — it is which /// of the tail pair the graph let run. The failure note comes from the DAG's /// first failing node, since the branch knows *that* it failed but not *why*. diff --git a/hive-c0re/src/job_queue/model.rs b/hive-c0re/src/job_queue/model.rs index be77e688..2f5e4573 100644 --- a/hive-c0re/src/job_queue/model.rs +++ b/hive-c0re/src/job_queue/model.rs @@ -78,36 +78,6 @@ pub enum NodeKind { /// First-spawn `nixos-container create` proper. Assumes the /// upstream `Provision` node already registered the agent in meta. Create { agent: String }, - /// `nixos-container destroy` plus the un-registration that follows it: - /// drop the agent from the coordinator's roster and clear its ephemeral - /// runtime dir. - /// - /// **Deliberately not in [`NodeKind::takes_container_down`]**, and that - /// is the design rather than an oversight. This node runs *downstream of - /// a `Stop`*, which already carries the flag honestly, so by the time it - /// claims there is nothing left to take down. A container still live here - /// is a real bug and must page someone — a `true` would absorb exactly - /// that signal, and the flag's whole asymmetry (see that method) is that - /// a wrong `true` silently swallows a crash. - DestroyContainer { agent: String }, - /// The `purge = true` half of a destroy: delete the agent's state - /// subvolume (via hive-priv, since a subvolume root defeats - /// `remove_dir_all`) plus its state and applied dirs. Its own node - /// because it is conditional — a plain destroy never inserts it — and - /// because it is the irreversible step, so it earns a distinct row in - /// the graph rather than hiding inside a bookkeeping tail. - PurgeState { agent: String }, - /// The post-destroy bookkeeping tail: meta sync, fail the agent's pending - /// approvals, drop the durable power intent, notify the manager, rescan - /// containers, re-emit the tombstone + schedule snapshots, resync - /// tmpfiles. Split from [`NodeKind::DestroyContainer`] for the same - /// reason [`NodeKind::RebuildBookkeeping`] is split from `Swap`: - /// dashboard visibility and retry granularity for work that is pure - /// store/meta bookkeeping and touches no container. - /// - /// `purge` only selects the wording of the approval-failure reason and - /// the manager notification; the destructive work is `PurgeState`'s. - DestroyBookkeeping { agent: String, purge: bool }, /// Meta flake lock bump. `sweep = false`: `meta::lock_update` /// (commit fused, under `META_LOCK`) with this node's own `inputs`; /// `sweep = true`: `meta::lock_update_hyperhive`, *non-fatal* (a @@ -371,9 +341,6 @@ impl NodeKind { NodeKind::RebuildBookkeeping { .. } => "rebuild_bookkeeping", NodeKind::Provision { .. } => "provision", NodeKind::Create { .. } => "create", - NodeKind::DestroyContainer { .. } => "destroy_container", - NodeKind::PurgeState { .. } => "purge_state", - NodeKind::DestroyBookkeeping { .. } => "destroy_bookkeeping", NodeKind::MetaLock { .. } => "meta_lock", NodeKind::Reconcile { .. } => "reconcile", NodeKind::Start { .. } => "start", @@ -411,9 +378,6 @@ impl NodeKind { | NodeKind::RebuildBookkeeping { agent } | NodeKind::Provision { agent } | NodeKind::Create { agent } - | NodeKind::DestroyContainer { agent } - | NodeKind::PurgeState { agent } - | NodeKind::DestroyBookkeeping { agent, .. } | NodeKind::Reconcile { agent } | NodeKind::Start { agent } | NodeKind::Stop { agent } @@ -471,12 +435,6 @@ impl NodeKind { // - `Create` / `Start` / `SetWanted{up}` bring a container UP. A // container disappearing *while starting* is a genuine crash and has // to keep reporting as one. - // - `DestroyContainer` looks like the most obvious `true` on this list - // and is the one that must stay `false`. It is edged downstream of a - // `Stop`, so the container is already down when it claims; the stop - // that the operator asked for is accounted for by the node that - // performs it. A container found alive at destroy time is a genuine - // bug, and a `true` here would suppress the alert that says so. // - `Reconcile` is a planner; it fans out `Start` / `Stop`, which carry // their own answer. // - `DeployWindow` brackets a deploy without itself stopping anything. diff --git a/hive-c0re/src/job_queue/templates.rs b/hive-c0re/src/job_queue/templates.rs index 94d7366a..ee67e388 100644 --- a/hive-c0re/src/job_queue/templates.rs +++ b/hive-c0re/src/job_queue/templates.rs @@ -474,59 +474,6 @@ pub fn spawn(builder: &JobBuilder, agent: &str, approval_id: i64) { resolve_approval_tails(builder, approval_id, provision); } -/// Teardown: `Stop` → `DestroyContainer` → (`PurgeState`) → `DestroyBookkeeping`. -/// -/// The chain is the point, not a decomposition for its own sake. Destroy used -/// to be a straight-line async fn with no queue node behind it, so nothing in -/// the graph could answer "is this container going down on purpose?" — which is -/// why an imperative crash-watch suppression guard existed at all. Reusing the -/// existing [`NodeKind::Stop`] answers it structurally: `Stop` already declares -/// `takes_container_down`, so the suppression is derived from the graph like -/// every other lifecycle op's. -/// -/// That also makes the precondition an edge rather than an assertion. -/// `DestroyContainer` runs only after `Stop` succeeded, so it operates on an -/// already-stopped container and carries `takes_container_down = false` -/// permanently — a container still alive at that point is a real bug and stays -/// loud instead of being absorbed by a flag. -/// -/// `Stop` is idempotent against an already-down container, so the common -/// "destroy something that isn't running" path costs nothing extra. -/// -/// `Stop` is the group root and holds the agent lease for the whole teardown; -/// the rest are `part_of` children that borrow it, so no other op can interleave -/// with a half-destroyed agent. `PurgeState` is inserted only when asked for — -/// the graph shows the irreversible step as its own row when it happens, and -/// omits it entirely when it doesn't. -pub fn destroy(builder: &JobBuilder, agent: &str, purge: bool) { - let a = || agent.to_owned(); - let stop = builder - .node(NodeKind::Stop { agent: a() }) - .needs(Resource::Agent(a())); - // `part_of` IS the ordering: a child runs once its parent reaches - // `Finishing`, and a node may not also declare a dep on its own parent - // (dep-scope validation rejects it — it would deadlock). So the - // "container is already stopped" precondition is the group edge itself, - // with no explicit `after_ok(stop)` to add. - let destroy = builder - .node(NodeKind::DestroyContainer { agent: a() }) - .part_of(stop); - // The bookkeeping tail hangs off the purge when there is one, so the - // irreversible delete lands before the meta sync that stops referencing it. - let last = if purge { - builder - .node(NodeKind::PurgeState { agent: a() }) - .part_of(stop) - .after_ok(destroy) - } else { - destroy - }; - let _tail = builder - .node(NodeKind::DestroyBookkeeping { agent: a(), purge }) - .part_of(stop) - .after_ok(last); -} - /// Perm change: commit the JSON file(s), then the rebuild subgraph so /// the updated `HIVE_TOOL_GROUPS` / `HIVE_CAPABILITIES` env var takes /// effect in the container. Group-roots are `WritePermFile` plus the rebuild diff --git a/hive-c0re/src/job_queue/tests.rs b/hive-c0re/src/job_queue/tests.rs index 4e27357a..4ac0a3f8 100644 --- a/hive-c0re/src/job_queue/tests.rs +++ b/hive-c0re/src/job_queue/tests.rs @@ -1714,97 +1714,6 @@ fn spawn_shape_provision_create_dropin_reconcile() { ); } -/// The destroy chain, and the reason it is a chain: `Stop` is reused so the -/// crash-watch answer comes from the node that actually stops the container. -/// -/// Asserting the *edges* is the point. `destroy_container` runs `after_ok` a -/// `stop`, which is what makes "the container is already down here" a -/// structural fact rather than a convention — see the companion test below for -/// why that matters. -#[test] -fn destroy_shape_stop_then_destroy_then_bookkeeping() { - let q = JobQueue::new(1); - insert(&q, |builder| { - templates::destroy(builder, "doomed", false); - }); - assert_eq!( - declared_shape(&q), - vec![ - row("stop", None, &[]), - // No explicit edge to `stop`: `part_of` already gates the child on - // its parent reaching `Finishing`, and declaring a dep on your own - // parent is rejected outright (it would deadlock). The precondition - // is the group membership. - row("destroy_container", Some("stop"), &[]), - row( - "destroy_bookkeeping", - Some("stop"), - &[("destroy_container", "done")] - ), - ] - ); -} - -/// `purge` inserts the irreversible delete as its own node, between the destroy -/// and the bookkeeping tail — so the meta sync that stops referencing the agent -/// runs *after* its trees are actually gone, and a purge is visibly distinct -/// from a plain destroy on the graph instead of being a hidden boolean. -#[test] -fn destroy_shape_purge_inserts_purge_state_before_the_tail() { - let q = JobQueue::new(1); - insert(&q, |builder| { - templates::destroy(builder, "doomed", true); - }); - assert_eq!( - declared_shape(&q), - vec![ - row("stop", None, &[]), - row("destroy_container", Some("stop"), &[]), - row( - "purge_state", - Some("stop"), - &[("destroy_container", "done")] - ), - row( - "destroy_bookkeeping", - Some("stop"), - &[("purge_state", "done")] - ), - ] - ); -} - -/// The counter-case to `rebuild_chain_nodes_suppress_crash_watch`, and the one -/// assertion in this file that exists to stop a *plausible* edit rather than a -/// wrong one. -/// -/// `destroy_container` is the most obvious candidate for `takes_container_down` -/// on the whole list and must stay `false`. It is edged downstream of a `Stop` -/// that already carries the flag, so the intentional stop is already accounted -/// for; a container still alive when this node claims is a genuine bug. Since a -/// wrong `true` **silently swallows a real crash** while a wrong `false` only -/// costs a spurious event, this is the asymmetry that has to be pinned. -#[test] -fn destroy_container_must_not_suppress_crash_watch() { - assert!( - !NodeKind::DestroyContainer { - agent: "a".to_owned() - } - .takes_container_down(), - "destroy_container runs after a Stop that already declared the \ - container is going down; claiming it again would suppress the alert \ - for a container found unexpectedly alive" - ); - // The upstream node is where the `true` lives — assert it here too, so the - // pair reads as one property and moving the flag breaks this test. - assert!( - NodeKind::Stop { - agent: "a".to_owned() - } - .takes_container_down() - ); -} - #[test] fn perm_change_shape_prefixes_rebuild_chain() { let q = JobQueue::new(1); diff --git a/hive-c0re/src/lifecycle/host_config.rs b/hive-c0re/src/lifecycle/host_config.rs index cedf84cf..8d5ac725 100644 --- a/hive-c0re/src/lifecycle/host_config.rs +++ b/hive-c0re/src/lifecycle/host_config.rs @@ -2,7 +2,7 @@ //! network isolation, forwarded credentials), the systemd resource-limits //! drop-in, and the `write_dropins` verb that re-applies both. -use std::path::{Path, PathBuf}; +use std::path::Path; use anyhow::{Context, Result}; use hive_priv_sock::{BindMount, CredentialMount}; @@ -71,19 +71,6 @@ async fn systemd_daemon_reload() -> Result<()> { /// inside the container. pub const CONTAINER_MANAGER_APPLIED_MOUNT: &str = "/applied"; -/// Host path behind every `/agents//config` mount: the **applied** -/// (deployed) repo, not the working clone at `agents//config`. That -/// clone is where a config change is staged, so it can hold a proposal -/// that is still under review or was rejected outright — mounting it shows -/// an agent a config which does not govern it. Both mounts (an agent's own -/// and a parent's view of a child's) go through here so they cannot drift. -/// -/// Never empty under a live container: `provision_container` runs -/// `setup_applied` before `create_only` makes the container at all. -fn config_bind_source(name: &str) -> PathBuf { - crate::paths::applied_dir(name) -} - /// Append bind flags for `child`'s state and config dirs into `binds`. /// See docs/persistence.md ("Parent access to child state") for what a /// parent may touch and why. Creates missing host-side directories so @@ -117,10 +104,8 @@ fn bind_child_agent_dirs(child: &str, binds: &mut Vec) { return; }; let child_root = crate::paths::agent_state_dir(&child); - for (sub, host, read_only) in [ - ("state", child_root.join("state"), false), - ("config", config_bind_source(child.as_str()), true), - ] { + for (sub, read_only) in [("state", false), ("config", true)] { + let host = child_root.join(sub); let _ = std::fs::create_dir_all(&host); binds.push(BindMount { host_path: host.to_string_lossy().into_owned(), @@ -264,9 +249,9 @@ async fn set_nspawn_flags( read_only: false, }); } - hive_types::Ident::parse(agent_name) + let agent_id = hive_types::Ident::parse(agent_name) .map_err(|e| anyhow::anyhow!("invalid agent name {agent_name:?}: {e}"))?; - let own_config = config_bind_source(agent_name); + let own_config = crate::paths::agent_state_dir(&agent_id).join("config"); std::fs::create_dir_all(&own_config) .with_context(|| format!("create {}", own_config.display()))?; binds.push(BindMount { @@ -396,27 +381,6 @@ mod tests { assert_eq!(paths, ["/agents/kiddo/state", "/agents/kiddo/config"]); } - /// The `config` mount names the **deployed** tree, not the working - /// clone the proposal is staged in. Asserted as "outside the child's - /// own dir" rather than by equality: the point is that the two are - /// different objects, which is what makes the mount unable to show a - /// config that was never approved. Equality with `applied_dir` would - /// restate the implementation and pass under any future relocation. - #[test] - fn child_config_mount_is_the_deployed_tree_not_the_working_clone() { - let working_clone = - crate::paths::agent_state_dir(&hive_types::Ident::parse("kiddo").expect("valid ident")); - let config = child_binds() - .into_iter() - .find(|b| b.container_path.ends_with("/config")) - .expect("a config bind"); - assert!( - !std::path::Path::new(&config.host_path).starts_with(&working_clone), - "config mount must not come from the child's working clone: {}", - config.host_path - ); - } - /// The regression this exists for. `harness` holds the child's own /// runtime material and was only ever mounted because one loop /// treated all three dirs alike — re-adding it to that loop is a diff --git a/hive-c0re/src/server.rs b/hive-c0re/src/server.rs index 753e67c3..eedb54d7 100644 --- a/hive-c0re/src/server.rs +++ b/hive-c0re/src/server.rs @@ -144,7 +144,7 @@ async fn dispatch(req: &HostRequest, coord: Arc) -> HostResponse { handle_start(&coord, &agents, &infra).await? } HostRequest::Destroy { name, purge } => { - actions::destroy(&coord, name.as_str(), *purge); + actions::destroy(&coord, name.as_str(), *purge).await?; HostResponse::success() } HostRequest::Rebuild { name } => { diff --git a/hive-c0re/src/stores/audit_log.rs b/hive-c0re/src/stores/audit_log.rs index e2928f2b..eb1cd377 100644 --- a/hive-c0re/src/stores/audit_log.rs +++ b/hive-c0re/src/stores/audit_log.rs @@ -27,7 +27,6 @@ use anyhow::{Context, Result}; use chrono::{DateTime, Utc}; use rusqlite::{Connection, params}; use serde::Serialize; -use utoipa::ToSchema; /// Process-singleton handle, set once at coordinator startup. Mirrors /// `build_logs::GLOBAL` — lets recording sites write without threading an @@ -81,7 +80,7 @@ impl AuditOutcome { } /// One audit row as returned to the dashboard. -#[derive(Debug, Clone, Serialize, ToSchema)] +#[derive(Debug, Clone, Serialize)] pub struct AuditEntry { pub id: i64, pub ts_unix: DateTime, diff --git a/hive-c0re/src/workers/crash_watch.rs b/hive-c0re/src/workers/crash_watch.rs index cc15e6f3..6ee46982 100644 --- a/hive-c0re/src/workers/crash_watch.rs +++ b/hive-c0re/src/workers/crash_watch.rs @@ -88,17 +88,17 @@ fn emit_crash_transitions(coord: &Coordinator, prev: &HashSet, current: // guard between two crash-watch polls. let recent = coord.recent_transient_within(RECENT_TRANSIENT_GRACE); for stopped in prev.difference(current) { - // One source: a running queue node that declared it takes the container - // down. There used to be a second — an imperative suppression guard for - // operations with no node behind them — and destroy becoming a DAG - // removed the last caller, so intent now has exactly one home. + // Two sources, because a container can go down on purpose either way: + // a running queue node that declared it takes the container down, or a + // no-node operation (destroy, migration) holding a suppression guard. // `any`, not "the" pill: an agent can have several running nodes at // once (a lease-exempt build alongside a lease-holding stop), and it // only takes one of them expecting the container down for this to be // a deliberate stop rather than a crash. let active = transients .get(stopped) - .map(|sts| sts.iter().any(|st| st.takes_container_down)); + .map(|sts| sts.iter().any(|st| st.takes_container_down)) + .or_else(|| coord.crash_watch_suppressed(stopped).then_some(true)); let recently_cleared = recent.get(stopped).copied(); if is_deliberate_stop(active, recently_cleared) { continue; diff --git a/hive-priv/Cargo.toml b/hive-priv/Cargo.toml index c63d98b2..a322d0f9 100644 --- a/hive-priv/Cargo.toml +++ b/hive-priv/Cargo.toml @@ -11,7 +11,6 @@ workspace = true anyhow.workspace = true hive-priv-sock.workspace = true libc.workspace = true -serde.workspace = true serde_json.workspace = true tokio.workspace = true tracing.workspace = true diff --git a/hive-priv/src/main.rs b/hive-priv/src/main.rs index 7abaf198..66d1f803 100644 --- a/hive-priv/src/main.rs +++ b/hive-priv/src/main.rs @@ -27,7 +27,6 @@ use hive_priv_sock::{ NetworkIsolation, PAUSED_MARKER_FILE, PRIV_SOCK, PrivEvent, PrivRequest, PrivResponse, PrivStream, PrivStreamLine, SIBLING_CONTAINERS, }; -use serde::Serialize; use tokio::io::{AsyncWriteExt, BufReader}; use tokio::net::unix::OwnedWriteHalf; use tokio::net::{UnixListener, UnixStream}; @@ -469,10 +468,7 @@ async fn exec( // when both `account` and `homeserver` are present; the account // suffix is already validated above. if let (Some(a), Some(hs)) = (account, homeserver) { - let meta = serde_json::to_string(&MatrixAccountSidecar { - homeserver: hs.as_str(), - }) - .context("serialize matrix account sidecar")?; + let meta = serde_json::json!({ "homeserver": hs }).to_string(); write_agent_state_file(agent_name, &format!("matrix-account-{a}.json"), &meta)?; } Ok(res) @@ -501,10 +497,7 @@ async fn exec( )?; // Sidecar carries the base URL — there's no host-side nix config // for extra forges, so this is the only place it's persisted. - let meta = serde_json::to_string(&ForgeSidecar { - base_url: base_url.as_str(), - }) - .context("serialize forge account sidecar")?; + let meta = serde_json::json!({ "base_url": base_url }).to_string(); write_agent_state_file(agent_name, &format!("forge-{label}.json"), &meta)?; Ok(res) } @@ -964,32 +957,6 @@ fn write_state_file_nofollow(dir: &Path, filename: &str, content: &str) -> Resul Ok(file) } -/// Sidecar written alongside an extra matrix account's token -/// (`matrix-account-.json`) so `hive-matrix-mcp` can auto-discover -/// the account's homeserver without a static `matrixAccounts` config -/// entry. Read side: `hive-matrix-mcp/src/accounts.rs`'s -/// `read_account_homeserver` (deliberately reads via a bare -/// `serde_json::Value` rather than this shape — that side treats a -/// malformed/missing sidecar as "skip this account" rather than an -/// error, so it stays loosely typed; this side is the one place the -/// file is written, so it gets the precise shape). -#[derive(Serialize)] -struct MatrixAccountSidecar<'a> { - homeserver: &'a str, -} - -/// Sidecar written alongside a dashboard-provisioned extra forge -/// account's token (`forge-