diff --git a/hive-ag3nt/src/events.rs b/hive-ag3nt/src/events.rs index bae95e30..8f0caea9 100644 --- a/hive-ag3nt/src/events.rs +++ b/hive-ag3nt/src/events.rs @@ -143,10 +143,7 @@ pub enum LiveEvent { /// compacting). `since_unix` matches `Bus::state_snapshot().1` /// so the client's elapsed-time ticker keeps progressing across /// SSE reconnects without drift. - TurnStateChanged { - state: TurnState, - since_unix: i64, - }, + TurnStateChanged { state: TurnState, since_unix: i64 }, } /// sqlite-backed event log. Wraps a `Connection` behind a `Mutex` so the @@ -284,10 +281,12 @@ impl TokenUsage { let model_usage = v.get("modelUsage")?; let map = model_usage.as_object()?; for (_model, stats) in map { - if let Some(w) = stats.get("contextWindow").and_then(serde_json::Value::as_u64) { - if w > 0 { - return Some(w); - } + if let Some(w) = stats + .get("contextWindow") + .and_then(serde_json::Value::as_u64) + && w > 0 + { + return Some(w); } } None @@ -353,18 +352,21 @@ pub fn context_window_tokens(model: &str) -> u64 { // Per-model env vars set by `hyperhive.contextWindowTokens` in Nix. for (key, val) in std::env::vars() { if let Some(suffix) = key.strip_prefix("HIVE_CONTEXT_WINDOW_TOKENS_") - && !suffix.is_empty() && m.contains(&suffix.to_ascii_lowercase()) - && let Ok(v) = val.trim().parse::() - && v > 0 { - return v; - } + && !suffix.is_empty() + && m.contains(&suffix.to_ascii_lowercase()) + && let Ok(v) = val.trim().parse::() + && v > 0 + { + return v; + } } // Global override (single value, any model). if let Ok(s) = std::env::var("HIVE_CONTEXT_WINDOW_TOKENS") && let Ok(v) = s.trim().parse::() - && v > 0 { - return v; - } + && v > 0 + { + return v; + } // Hard fallback for dev/test outside NixOS where env vars aren't set. 200_000 } @@ -450,9 +452,10 @@ impl Bus { // persisted runtime override > compiled-in DEFAULT_MODEL. // The nix config always wins on rebuild; the persisted file is kept // for within-session tracking only (see persist_model / set_model). - let initial_model = configured_model() - .map(str::to_owned) - .unwrap_or_else(|| load_model().unwrap_or_else(|| DEFAULT_MODEL.to_owned())); + let initial_model = configured_model().map_or_else( + || load_model().unwrap_or_else(|| DEFAULT_MODEL.to_owned()), + str::to_owned, + ); // Restore rate_limited from the sentinel file — if the harness // crashed while parked, we should still show the right status on // cold load until the next turn clears it. @@ -560,7 +563,8 @@ impl Bus { pub fn record_turn_usage(&self, ctx: TokenUsage, cost: TokenUsage) { *self.last_ctx_usage.lock().unwrap() = Some(ctx); *self.last_cost_usage.lock().unwrap() = Some(cost); - self.last_turn_ended_unix.store(now_unix(), Ordering::Relaxed); + self.last_turn_ended_unix + .store(now_unix(), Ordering::Relaxed); self.emit(LiveEvent::TokenUsageChanged { ctx, cost }); } diff --git a/hive-ag3nt/src/forge_notify.rs b/hive-ag3nt/src/forge_notify.rs index 92b40b59..cba2cd5e 100644 --- a/hive-ag3nt/src/forge_notify.rs +++ b/hive-ag3nt/src/forge_notify.rs @@ -162,11 +162,7 @@ pub async fn run(socket: PathBuf, is_manager: bool) { /// Fetch a JSON value from a URL using the agent's forge token. Returns /// `None` on any HTTP or parse error (best-effort enrichment). -async fn fetch_json( - client: &reqwest::Client, - url: &str, - token: &str, -) -> Option { +async fn fetch_json(client: &reqwest::Client, url: &str, token: &str) -> Option { let resp = client .get(url) .header("Authorization", format!("token {token}")) @@ -207,7 +203,7 @@ fn notif_type_label(t: &str) -> &str { /// (`heading\n===`) are not handled here — rarer in practice and /// would need multi-line lookahead; revisit if it actually shows up. /// -/// **ATX shape strictly:** CommonMark requires a space (or end-of-line) +/// **ATX shape strictly:** `CommonMark` requires a space (or end-of-line) /// after the 1-6 leading `#`s to count as an ATX heading. Lines like /// `#tag`, `#123`, `#!/bin/bash` are NOT headings — passing them through /// untouched avoids the cosmetic noise argus flagged on PR #518 (`\#tag` @@ -236,7 +232,7 @@ fn escape_md_headings(body: &str) -> String { out } -/// Strict CommonMark ATX-heading detector: 1-6 leading `#`s followed +/// Strict `CommonMark` ATX-heading detector: 1-6 leading `#`s followed /// by either a space, tab, or end-of-line. Anything tighter (`#tag`, /// `#123`) is a non-heading line that the renderer will not promote. fn is_atx_heading(line: &str) -> bool { @@ -244,11 +240,10 @@ fn is_atx_heading(line: &str) -> bool { if !(1..=6).contains(&hashes) { return false; } - match line.as_bytes().get(hashes) { - None => true, // bare `#` / `##` / ... on its own line - Some(b' ') | Some(b'\t') => true, // proper ATX with space/tab after #s - _ => false, // `#tag` / `#123` — not a heading - } + // Bare `#` / `##` / ... on its own line, or proper ATX with a + // space/tab after the run of `#`s; anything else (`#tag` / `#123`) + // is not a heading. + matches!(line.as_bytes().get(hashes), None | Some(b' ' | b'\t')) } fn truncate(s: &str, max: usize) -> String { @@ -390,7 +385,9 @@ async fn format_notification( // API URLs for fetching content let subject_api_url = notif["subject"]["url"].as_str().unwrap_or(""); - let comment_api_url = notif["subject"]["latest_comment_url"].as_str().unwrap_or(""); + let comment_api_url = notif["subject"]["latest_comment_url"] + .as_str() + .unwrap_or(""); let comment_html_url = notif["subject"]["latest_comment_html_url"] .as_str() .unwrap_or(""); @@ -411,9 +408,27 @@ async fn format_notification( // by creation/state-change of the subject itself. let has_comment = !comment_api_url.is_empty() && comment_api_url != subject_api_url; - let meta = NotifMeta { title, notif_type, html_url, num, repo, meta_suffix, reason, subject, is_pr }; + let meta = NotifMeta { + title, + notif_type, + html_url, + num, + repo, + meta_suffix, + reason, + subject, + is_pr, + }; if has_comment { - format_comment_notification(client, token, &meta, comment_api_url, comment_html_url, own_login).await + format_comment_notification( + client, + token, + &meta, + comment_api_url, + comment_html_url, + own_login, + ) + .await } else { format_state_change_notification(notif, &meta, own_login) } @@ -454,16 +469,28 @@ fn build_meta_suffix(subject: Option<&serde_json::Value>, is_pr: bool, reason: & .and_then(|s| s["requested_reviewers"].as_array()) .map(|arr| arr.iter().filter_map(|r| r["login"].as_str()).collect()) .unwrap_or_default(); - if reviewers.is_empty() { None } else { Some(format!("reviewer: {}", reviewers.join(", "))) } + if reviewers.is_empty() { + None + } else { + Some(format!("reviewer: {}", reviewers.join(", "))) + } } else { None }; // Always include reason so multiple notifications for the same event // (each with a different Forgejo reason) are distinguishable (closes #110). - let reason_line = if reason.is_empty() { None } else { Some(format!("reason: {reason}")) }; + let reason_line = if reason.is_empty() { + None + } else { + Some(format!("reason: {reason}")) + }; let mut out = format!("\n{assignee_line}"); - if let Some(r) = reviewer_line { write!(out, "\n{r}").ok(); } - if let Some(r) = reason_line { write!(out, "\n{r}").ok(); } + if let Some(r) = reviewer_line { + write!(out, "\n{r}").ok(); + } + if let Some(r) = reason_line { + write!(out, "\n{r}").ok(); + } out } @@ -504,9 +531,24 @@ async fn format_comment_notification( .and_then(|c| c["state"].as_str()) .and_then(review_state_label); - let url = if comment_html_url.is_empty() { meta.html_url } else { comment_html_url }; - let author = if actor_login.is_empty() { "?" } else { actor_login }; - let NotifMeta { title, notif_type, num, repo, meta_suffix, .. } = meta; + let url = if comment_html_url.is_empty() { + meta.html_url + } else { + comment_html_url + }; + let author = if actor_login.is_empty() { + "?" + } else { + actor_login + }; + let NotifMeta { + title, + notif_type, + num, + repo, + meta_suffix, + .. + } = meta; // Truncate the raw body first so the mention-overflow diff compares // like-for-like (escape_md_headings rewrites `# foo` to `\# foo`, so @@ -574,7 +616,17 @@ fn format_state_change_notification( return None; } - let NotifMeta { title, notif_type, html_url, num, repo, meta_suffix, reason: _, subject, is_pr } = meta; + let NotifMeta { + title, + notif_type, + html_url, + num, + repo, + meta_suffix, + reason: _, + subject, + is_pr, + } = meta; let label = notif_type_label(notif_type); let kind = match notif_state { "merged" => format!("{label} merged{num}{repo}"), @@ -630,6 +682,12 @@ fn format_state_change_notification( } #[allow(clippy::too_many_arguments)] +#[allow( + clippy::too_many_lines, + reason = "single-pass notification poll loop — split would obscure the \ + sequential 'fetch / classify / dispatch' rhythm and add helper \ + functions for state shared across all three phases" +)] async fn poll_once( client: &reqwest::Client, forge_url: &str, @@ -672,10 +730,15 @@ async fn poll_once( return; } - debug!(count = notifications.len(), "forge_notify: delivering notifications"); + debug!( + count = notifications.len(), + "forge_notify: delivering notifications" + ); for notif in ¬ifications { - let Some(id) = notif["id"].as_u64() else { continue }; + let Some(id) = notif["id"].as_u64() else { + continue; + }; // Reason drop-list: suppress noisy reasons (subscribed/participating). // null/unknown reasons pass through — directed signals are never @@ -733,28 +796,30 @@ async fn poll_once( // when HIVE_FORGE_KEEP_SUBSCRIPTIONS=1 — triage and other firehose // consumers set this to retain broad repo visibility. let reason = notif["reason"].as_str().unwrap_or(""); - if !keep_subscriptions && reason == "subscribed" + if !keep_subscriptions + && reason == "subscribed" && let Some(repo) = notif["repository"]["full_name"].as_str() - && !unsubbed_repos.contains(repo) { - let unsub_url = format!("{forge_url}/api/v1/repos/{repo}/subscription"); - match client - .delete(&unsub_url) - .header("Authorization", format!("token {token}")) - .send() - .await - { - Ok(r) if r.status().is_success() || r.status().as_u16() == 404 => { - debug!(%repo, "forge_notify: unsubscribed from repo watch"); - unsubbed_repos.insert(repo.to_owned()); - } - Ok(r) => { - debug!(%repo, status = %r.status(), "forge_notify: unsub non-2xx (ignored)"); - } - Err(e) => { - debug!(%repo, error = ?e, "forge_notify: unsub request failed (ignored)"); - } - } + && !unsubbed_repos.contains(repo) + { + let unsub_url = format!("{forge_url}/api/v1/repos/{repo}/subscription"); + match client + .delete(&unsub_url) + .header("Authorization", format!("token {token}")) + .send() + .await + { + Ok(r) if r.status().is_success() || r.status().as_u16() == 404 => { + debug!(%repo, "forge_notify: unsubscribed from repo watch"); + unsubbed_repos.insert(repo.to_owned()); } + Ok(r) => { + debug!(%repo, status = %r.status(), "forge_notify: unsub non-2xx (ignored)"); + } + Err(e) => { + debug!(%repo, error = ?e, "forge_notify: unsub request failed (ignored)"); + } + } + } } } diff --git a/hive-ag3nt/src/mcp.rs b/hive-ag3nt/src/mcp.rs index 4cbafcde..ca94aa6c 100644 --- a/hive-ag3nt/src/mcp.rs +++ b/hive-ag3nt/src/mcp.rs @@ -47,7 +47,7 @@ pub enum SocketReply { Recent(Vec), Logs(String), /// `list_schedules` result — used by the manager surface only; - /// AgentResponse has no equivalent variant. + /// `AgentResponse` has no equivalent variant. Schedules(Vec), LooseEnds(Vec), PendingRemindersCount(u64), @@ -185,8 +185,7 @@ pub fn format_recv(resp: Result) -> String { /// resurfaced by `RequeueInflight` on this session's boot. Same /// string surfaces in the wake prompt (see the bin loops) and the /// in-turn `recv` tool result so claude sees the warning either way. -pub const REDELIVERY_HINT: &str = - "[redelivered after harness restart — may already be handled]\n"; +pub const REDELIVERY_HINT: &str = "[redelivered after harness restart — may already be handled]\n"; /// Format helper for `get_loose_ends`: renders a short bulleted list /// of pending approvals + questions + reminders. Empty list collapses @@ -307,17 +306,32 @@ pub fn format_agent_meta(resp: Result) -> String { match status_text { None => out.push_str("\nstatus: "), Some(s) => { + use std::fmt::Write as _; let age = status_set_at.and_then(|ts| { let now = std::time::SystemTime::now() .duration_since(std::time::UNIX_EPOCH) .ok()? .as_secs(); - let secs = now.saturating_sub(ts as u64); + // `ts` is a unix epoch second the agent itself + // sourced from `SystemTime` — always positive + // in normal operation. Clamp the negative + // (clock-skew) edge to 0 before the unsigned + // cast so the cast loses no real precision. + let ts_secs = u64::try_from(ts).unwrap_or(0); + let secs = now.saturating_sub(ts_secs); Some(format_age_secs(secs)) }); + // `write!` into the buffer instead of `push_str(&format!(…))` — + // avoids the intermediate allocation clippy::format_push_string + // flags. The infallible `String` writer makes this safe to + // `let _ =`-ignore. match age { - Some(a) => out.push_str(&format!("\nstatus: {s} (set {a} ago)")), - None => out.push_str(&format!("\nstatus: {s}")), + Some(a) => { + let _ = write!(out, "\nstatus: {s} (set {a} ago)"); + } + None => { + let _ = write!(out, "\nstatus: {s}"); + } } } } @@ -626,9 +640,13 @@ impl AgentServer { )] async fn set_status(&self, Parameters(args): Parameters) -> String { run_tool_envelope("set_status", args.text.clone(), async move { - let (resp, retries) = - self.dispatch(hive_sh4re::AgentRequest::SetStatus { text: args.text }).await; - annotate_retries(format_ack(resp, "set_status", "status updated".to_owned()), retries) + let (resp, retries) = self + .dispatch(hive_sh4re::AgentRequest::SetStatus { text: args.text }) + .await; + annotate_retries( + format_ack(resp, "set_status", "status updated".to_owned()), + retries, + ) }) .await } @@ -644,10 +662,7 @@ impl AgentServer { where the system-prompt label could be stale. Status reads `` when the \ target has never called `set_status` or has cleared it." )] - async fn get_agent_meta( - &self, - Parameters(args): Parameters, - ) -> String { + async fn get_agent_meta(&self, Parameters(args): Parameters) -> String { let log = args.name.clone().unwrap_or_else(|| "".to_owned()); run_tool_envelope("get_agent_meta", log, async move { let (resp, retries) = self @@ -679,7 +694,11 @@ impl AgentServer { .dispatch(hive_sh4re::AgentRequest::CancelLooseEnd { kind, id }) .await; annotate_retries( - format_ack(resp, "cancel_loose_end", format!("cancelled {kind_label} {id}")), + format_ack( + resp, + "cancel_loose_end", + format!("cancelled {kind_label} {id}"), + ), retries, ) }) @@ -721,7 +740,10 @@ impl AgentServer { file_path: args.file_path, }) .await; - annotate_retries(format_ack(resp, "remind", "reminder scheduled".to_string()), retries) + annotate_retries( + format_ack(resp, "remind", "reminder scheduled".to_string()), + retries, + ) }) .await } @@ -949,7 +971,7 @@ pub struct RequestSchedulePromptArgs { /// `None` / absent = one-shot. `Some(n > 0)` = recurring every /// `n` seconds. The worker clamps catch-up so a long downtime /// fires ONCE on resume (skipped-cycle count surfaces in the - /// per-target last_result), not N delayed pulses in a row. + /// per-target `last_result`), not N delayed pulses in a row. #[serde(default)] pub interval_seconds: Option, /// Optional description shown on the dashboard approval card + @@ -1301,10 +1323,7 @@ impl ManagerServer { Authorization mirrors `cancel_schedule`: you can fire your own schedules + any \ owned by a sub-agent in your subtree per topology.json." )] - async fn fire_schedule_now( - &self, - Parameters(args): Parameters, - ) -> String { + async fn fire_schedule_now(&self, Parameters(args): Parameters) -> String { let log = format!("{args:?}"); run_tool_envelope("fire_schedule_now", log, async move { let id = args.id; @@ -1337,7 +1356,10 @@ impl ManagerServer { targets: args.targets, }) .await; - annotate_retries(format_ack(resp, "cancel_schedule", format!("cancelled #{id}")), retries) + annotate_retries( + format_ack(resp, "cancel_schedule", format!("cancelled #{id}")), + retries, + ) }) .await } @@ -1394,7 +1416,9 @@ impl ManagerServer { )] async fn list_schedules(&self) -> String { run_tool_envelope("list_schedules", String::new(), async move { - let (resp, retries) = self.dispatch(hive_sh4re::ManagerRequest::ListSchedules).await; + let (resp, retries) = self + .dispatch(hive_sh4re::ManagerRequest::ListSchedules) + .await; let body = match resp { Ok(SocketReply::Schedules(schedules)) => serde_json::to_string(&schedules) .unwrap_or_else(|e| format!("list_schedules: serialise: {e:#}")), @@ -1541,7 +1565,10 @@ impl ManagerServer { file_path: args.file_path, }) .await; - annotate_retries(format_ack(resp, "remind", "reminder scheduled".to_string()), retries) + annotate_retries( + format_ack(resp, "remind", "reminder scheduled".to_string()), + retries, + ) }) .await } @@ -1573,9 +1600,13 @@ impl ManagerServer { )] async fn set_status(&self, Parameters(args): Parameters) -> String { run_tool_envelope("set_status", args.text.clone(), async move { - let (resp, retries) = - self.dispatch(hive_sh4re::ManagerRequest::SetStatus { text: args.text }).await; - annotate_retries(format_ack(resp, "set_status", "status updated".to_owned()), retries) + let (resp, retries) = self + .dispatch(hive_sh4re::ManagerRequest::SetStatus { text: args.text }) + .await; + annotate_retries( + format_ack(resp, "set_status", "status updated".to_owned()), + retries, + ) }) .await } @@ -1590,10 +1621,7 @@ impl ManagerServer { drift across renames. Status reads `` when the target has never called \ `set_status` or has cleared it." )] - async fn get_agent_meta( - &self, - Parameters(args): Parameters, - ) -> String { + async fn get_agent_meta(&self, Parameters(args): Parameters) -> String { let log = args.name.clone().unwrap_or_else(|| "".to_owned()); run_tool_envelope("get_agent_meta", log, async move { let (resp, retries) = self @@ -1628,7 +1656,11 @@ impl ManagerServer { .dispatch(hive_sh4re::ManagerRequest::CancelLooseEnd { kind, id }) .await; annotate_retries( - format_ack(resp, "cancel_loose_end", format!("cancelled {kind_label} {id}")), + format_ack( + resp, + "cancel_loose_end", + format!("cancelled {kind_label} {id}"), + ), retries, ) }) @@ -1800,10 +1832,7 @@ pub fn allowed_tools_arg(flavor: Flavor) -> String { if patterns.is_empty() { vec!["Bash".to_owned()] } else { - patterns - .into_iter() - .map(|p| format!("Bash({p})")) - .collect() + patterns.into_iter().map(|p| format!("Bash({p})")).collect() } } else { vec![(*s).to_owned()] diff --git a/hive-ag3nt/src/turn.rs b/hive-ag3nt/src/turn.rs index c0b591b8..28f56205 100644 --- a/hive-ag3nt/src/turn.rs +++ b/hive-ag3nt/src/turn.rs @@ -235,6 +235,7 @@ fn effective_context_window(bus: &Bus) -> u64 { /// Resolve the auto-reset watermark. Priority order: /// 1. `HIVE_AUTO_RESET_WATERMARK_TOKENS` env var (explicit override). /// 2. 50% of `effective_context_window(bus)`. +/// /// `0` disables auto-reset entirely. fn auto_reset_watermark_tokens(bus: &Bus) -> u64 { if let Some(v) = std::env::var("HIVE_AUTO_RESET_WATERMARK_TOKENS") @@ -259,6 +260,7 @@ fn cache_ttl_secs() -> u64 { /// Resolve the proactive-compaction watermark. Priority order: /// 1. `HIVE_COMPACT_WATERMARK_TOKENS` env var (explicit override). /// 2. 75% of `effective_context_window(bus)`. +/// /// `0` disables proactive compaction (reactive path still applies). fn compact_watermark_tokens(bus: &Bus) -> u64 { if let Some(v) = std::env::var("HIVE_COMPACT_WATERMARK_TOKENS") @@ -299,9 +301,7 @@ pub async fn drive_turn(prompt: &str, files: &TurnFiles, bus: &Bus) -> TurnOutco // /compact itself would be absurd recursion; bubble it up as // a normal failure path. match compact_session(files, bus).await { - TurnOutcome::Ok | TurnOutcome::Compacted => { - run_turn(prompt, files, bus).await - } + TurnOutcome::Ok | TurnOutcome::Compacted => run_turn(prompt, files, bus).await, other => return other, } } @@ -315,10 +315,9 @@ pub async fn drive_turn(prompt: &str, files: &TurnFiles, bus: &Bus) -> TurnOutco // turn overflows into the reactive path. Best-effort — never changes // the outcome of the turn that already succeeded, but records it as // `Compacted` so turn stats can distinguish it from a plain `Ok`. - if matches!(outcome, TurnOutcome::Ok) - && maybe_checkpoint_and_compact(files, bus).await { - return TurnOutcome::Compacted; - } + if matches!(outcome, TurnOutcome::Ok) && maybe_checkpoint_and_compact(files, bus).await { + return TurnOutcome::Compacted; + } outcome } @@ -707,53 +706,48 @@ async fn run_claude(prompt: &str, files: &TurnFiles, bus: &Bus) -> Result<(bool, if AUTH_FAIL_MARKERS.iter().any(|m| line.contains(m)) { auth_out.store(true, Ordering::Relaxed); } - match serde_json::from_str::(&line) { - Ok(v) => { - // Rate-limit detection: only fire on JSON `error` events, - // not on arbitrary text content. An agent discussing a past - // rate limit in its response would otherwise trigger a false - // positive (the full conversation flows through stdout as - // stream-json, so any text the model outputs is visible here). - if v.get("type").and_then(|t| t.as_str()) == Some("error") { - let raw = v.to_string(); - if RATE_LIMIT_MARKERS.iter().any(|m| raw.contains(m)) { - rate_out.store(true, Ordering::Relaxed); - } - } - if let Some(u) = crate::events::TokenUsage::from_assistant_event(&v) { - last_inference = Some(u); - } - if let Some(cost) = crate::events::TokenUsage::from_stream_event(&v) { - // Fallback to `cost` if the turn somehow produced - // a result without any assistant event — keeps the - // ctx badge from going stale on a degenerate turn. - let ctx = last_inference.unwrap_or(cost); - bus_out.record_turn_usage(ctx, cost); - } - // Seed the API-reported context-window from the result - // event's `modelUsage.*.contextWindow` field. This is - // the authoritative per-inference active window used for - // compaction watermarks — it reflects what the model - // actually enforces, which may differ from the Nix - // config (e.g. 200k active window on a 1M cache model). - if let Some(w) = - crate::events::TokenUsage::context_window_from_result_event(&v) - { - bus_out.set_api_context_window(w); - } - bus_out.observe_stream(&v); - bus_out.emit(LiveEvent::Stream(v)); - } - Err(_) => { - // Non-JSON stdout: raw text check is fine here since these - // are claude CLI messages, not conversation content. - if RATE_LIMIT_MARKERS.iter().any(|m| line.contains(m)) { + if let Ok(v) = serde_json::from_str::(&line) { + // Rate-limit detection: only fire on JSON `error` events, + // not on arbitrary text content. An agent discussing a past + // rate limit in its response would otherwise trigger a false + // positive (the full conversation flows through stdout as + // stream-json, so any text the model outputs is visible here). + if v.get("type").and_then(|t| t.as_str()) == Some("error") { + let raw = v.to_string(); + if RATE_LIMIT_MARKERS.iter().any(|m| raw.contains(m)) { rate_out.store(true, Ordering::Relaxed); } - bus_out.emit(LiveEvent::Note { - text: format!("(non-json) {line}"), - }); } + if let Some(u) = crate::events::TokenUsage::from_assistant_event(&v) { + last_inference = Some(u); + } + if let Some(cost) = crate::events::TokenUsage::from_stream_event(&v) { + // Fallback to `cost` if the turn somehow produced + // a result without any assistant event — keeps the + // ctx badge from going stale on a degenerate turn. + let ctx = last_inference.unwrap_or(cost); + bus_out.record_turn_usage(ctx, cost); + } + // Seed the API-reported context-window from the result + // event's `modelUsage.*.contextWindow` field. This is + // the authoritative per-inference active window used for + // compaction watermarks — it reflects what the model + // actually enforces, which may differ from the Nix + // config (e.g. 200k active window on a 1M cache model). + if let Some(w) = crate::events::TokenUsage::context_window_from_result_event(&v) { + bus_out.set_api_context_window(w); + } + bus_out.observe_stream(&v); + bus_out.emit(LiveEvent::Stream(v)); + } else { + // Non-JSON stdout: raw text check is fine here since these + // are claude CLI messages, not conversation content. + if RATE_LIMIT_MARKERS.iter().any(|m| line.contains(m)) { + rate_out.store(true, Ordering::Relaxed); + } + bus_out.emit(LiveEvent::Note { + text: format!("(non-json) {line}"), + }); } } }); diff --git a/hive-c0re/src/actions.rs b/hive-c0re/src/actions.rs index 76998cf1..3ad26267 100644 --- a/hive-c0re/src/actions.rs +++ b/hive-c0re/src/actions.rs @@ -151,21 +151,18 @@ async fn run_approval_schedule_prompt( approval: hive_sh4re::Approval, ) -> Result<()> { let result: Result<()> = async { - let payload: hive_sh4re::SchedulePromptPayload = - serde_json::from_str(&approval.commit_ref) - .context("decode SchedulePromptPayload from approval.commit_ref")?; + let payload: hive_sh4re::SchedulePromptPayload = serde_json::from_str(&approval.commit_ref) + .context("decode SchedulePromptPayload from approval.commit_ref")?; coord .scheduled_prompts - .submit(crate::scheduled_prompts::NewSchedule { + .submit(&crate::scheduled_prompts::NewSchedule { owner: approval.agent.clone(), targets: payload.targets, body: payload.body, first_fire_at_unix: payload.first_fire_at_unix, interval_seconds: payload.interval_seconds, description: payload.description, - source: crate::scheduled_prompts::ScheduleSource::Approval { - id: approval.id, - }, + source: crate::scheduled_prompts::ScheduleSource::Approval { id: approval.id }, }) .map(|_| ()) .context("insert scheduled prompt") @@ -290,9 +287,10 @@ async fn forge_after_first_spawn(coord: &Arc, agent: &str) { tracing::warn!(%agent, error = ?e, "forge: ensure_config_repo after first spawn failed"); } if let Some(core_token) = crate::forge::core_token() - && let Err(e) = crate::forge::meta_read_access(agent, &core_token).await { - tracing::warn!(%agent, error = ?e, "forge: meta_read_access after first spawn failed"); - } + && let Err(e) = crate::forge::meta_read_access(agent, &core_token).await + { + tracing::warn!(%agent, error = ?e, "forge: meta_read_access after first spawn failed"); + } if let Err(e) = crate::forge::ensure_meta_remote(agent).await { tracing::warn!(%agent, error = ?e, "forge: ensure_meta_remote after first spawn failed"); } @@ -466,7 +464,7 @@ async fn run_apply_commit( Err(anyhow::anyhow!("read applied/main: {e:#}")), None, is_first_spawn, - ) + ); } }; @@ -521,8 +519,7 @@ async fn run_apply_commit( Ok(a) => a, Err(e) => { let _ = - lifecycle::git_update_ref(applied_dir, "refs/heads/main", &prev_main_sha) - .await; + lifecycle::git_update_ref(applied_dir, "refs/heads/main", &prev_main_sha).await; let _ = lifecycle::git_read_tree_reset(applied_dir, "refs/heads/main").await; return ( Err(anyhow::anyhow!("agents_for_meta_listing_with: {e:#}")), @@ -540,8 +537,7 @@ async fn run_apply_commit( ) .await { - let _ = - lifecycle::git_update_ref(applied_dir, "refs/heads/main", &prev_main_sha).await; + let _ = lifecycle::git_update_ref(applied_dir, "refs/heads/main", &prev_main_sha).await; let _ = lifecycle::git_read_tree_reset(applied_dir, "refs/heads/main").await; return ( Err(anyhow::anyhow!("meta sync_agents for first spawn: {e:#}")), diff --git a/hive-c0re/src/approvals.rs b/hive-c0re/src/approvals.rs index 957fff1f..df9979fd 100644 --- a/hive-c0re/src/approvals.rs +++ b/hive-c0re/src/approvals.rs @@ -251,9 +251,11 @@ impl Approvals { /// kind / agent / sha. Errors if the approval isn't pending — once /// it's approved/denied/failed/cancelled, the resolution is final. pub fn mark_cancelled(&self, id: i64, canceller: &str) -> Result { - let mut conn = self.conn.lock().unwrap(); - let tx = conn.transaction()?; - let row: Option<( + // Row-shape alias for the SELECT below so we don't trip + // clippy::type_complexity. Order matches the SELECT projection: + // agent, kind, commit_ref, requested_at, status, fetched_sha, + // description. + type CancelLookupRow = ( String, String, String, @@ -261,7 +263,10 @@ impl Approvals { String, Option, Option, - )> = tx + ); + let mut conn = self.conn.lock().unwrap(); + let tx = conn.transaction()?; + let row: Option = tx .query_row( "SELECT agent, kind, commit_ref, requested_at, status, fetched_sha, description FROM approvals WHERE id = ?1", @@ -326,9 +331,7 @@ impl Approvals { /// bad row used to make `pending()` / `recent_resolved()` error out /// wholesale — the dashboard then rendered an empty approvals queue /// (issue #160, an unhandled `init_config` kind poisoning every read). -fn collect_lenient( - rows: impl Iterator>, -) -> Vec { +fn collect_lenient(rows: impl Iterator>) -> Vec { rows.filter_map(|r| match r { Ok(a) => Some(a), Err(e) => { @@ -467,7 +470,12 @@ mod tests { // status + a "cancelled by " note. let (_dir, _path, db) = open_temp(); let id = db - .submit_kind("bitburner", ApprovalKind::ApplyCommit, "cafef00d", Some("test")) + .submit_kind( + "bitburner", + ApprovalKind::ApplyCommit, + "cafef00d", + Some("test"), + ) .unwrap(); let row = db.mark_cancelled(id, "manager").expect("cancel"); assert_eq!(row.id, id); diff --git a/hive-c0re/src/auto_update.rs b/hive-c0re/src/auto_update.rs index 31799898..dbf75e86 100644 --- a/hive-c0re/src/auto_update.rs +++ b/hive-c0re/src/auto_update.rs @@ -65,7 +65,7 @@ pub fn agent_config_pending(name: &str, deployed_sha: Option<&str>) -> bool { /// can't diverge. /// /// `queue_entry_id` is `Some(id)` when the rebuild was dispatched from -/// the rebuild_queue worker (lets the function annotate its phase via +/// the `rebuild_queue` worker (lets the function annotate its phase via /// `coord.set_queue_step`) and `None` when called directly (e.g. the /// manager-migration nudge in `ensure_manager`). pub async fn rebuild_agent( @@ -220,7 +220,10 @@ pub async fn run(coord: Arc) -> Result<()> { let _current_rev = current_flake_rev(&coord.hyperhive_flake).unwrap_or_default(); - tracing::info!(agents = containers.len(), "auto-update: queueing all on startup"); + tracing::info!( + agents = containers.len(), + "auto-update: queueing all on startup" + ); for container in containers { let logical = if container == MANAGER_NAME { Some(MANAGER_NAME.to_owned()) diff --git a/hive-c0re/src/crash_watch.rs b/hive-c0re/src/crash_watch.rs index fb7fd755..1c844d4d 100644 --- a/hive-c0re/src/crash_watch.rs +++ b/hive-c0re/src/crash_watch.rs @@ -139,6 +139,46 @@ fn is_deliberate_stop( active.is_some_and(is_op_kind) || recently_cleared.is_some_and(is_op_kind) } +fn emit_login_transitions( + coord: &Coordinator, + prev: &HashSet, + current: &HashSet, + sub_agents: &[String], + prev_sub_agents: &HashSet, +) { + for agent in current.difference(prev) { + tracing::info!(%agent, "agent logged in"); + coord.notify_manager(&hive_sh4re::HelperEvent::LoggedIn { + agent: agent.clone(), + }); + } + // Detect transitions into "needs login": an agent that was previously + // logged-in goes unsigned (credentials deleted), OR a brand-new agent + // appears without a session. + // + // prev_needs uses prev_sub_agents (the agent set from the last tick) so + // that a newly-spawned agent — which does not appear in prev_sub_agents — + // is absent from prev_needs even though it's not in prev_logged_in. + // Without this, new agents land in both prev_needs and current_needs and + // the set difference is empty, silently dropping the event. + let prev_needs: HashSet<&str> = prev_sub_agents + .iter() + .map(String::as_str) + .filter(|n| !prev.contains(*n)) + .collect(); + let current_needs: HashSet<&str> = sub_agents + .iter() + .map(String::as_str) + .filter(|n| !current.contains(*n)) + .collect(); + for agent in current_needs.difference(&prev_needs) { + tracing::info!(%agent, "agent needs login"); + coord.notify_manager(&hive_sh4re::HelperEvent::NeedsLogin { + agent: (*agent).to_owned(), + }); + } +} + #[cfg(test)] mod tests { use super::*; @@ -186,44 +226,3 @@ mod tests { } } } - -fn emit_login_transitions( - coord: &Coordinator, - prev: &HashSet, - current: &HashSet, - sub_agents: &[String], - prev_sub_agents: &HashSet, -) { - for agent in current.difference(prev) { - tracing::info!(%agent, "agent logged in"); - coord.notify_manager(&hive_sh4re::HelperEvent::LoggedIn { - agent: agent.clone(), - }); - } - // Detect transitions into "needs login": an agent that was previously - // logged-in goes unsigned (credentials deleted), OR a brand-new agent - // appears without a session. - // - // prev_needs uses prev_sub_agents (the agent set from the last tick) so - // that a newly-spawned agent — which does not appear in prev_sub_agents — - // is absent from prev_needs even though it's not in prev_logged_in. - // Without this, new agents land in both prev_needs and current_needs and - // the set difference is empty, silently dropping the event. - let prev_needs: HashSet<&str> = prev_sub_agents - .iter() - .map(String::as_str) - .filter(|n| !prev.contains(*n)) - .collect(); - let current_needs: HashSet<&str> = sub_agents - .iter() - .map(String::as_str) - .filter(|n| !current.contains(*n)) - .collect(); - for agent in current_needs.difference(&prev_needs) { - tracing::info!(%agent, "agent needs login"); - coord.notify_manager(&hive_sh4re::HelperEvent::NeedsLogin { - agent: (*agent).to_owned(), - }); - } -} - diff --git a/hive-c0re/src/dashboard.rs b/hive-c0re/src/dashboard.rs index 0a79bf11..39b284a0 100644 --- a/hive-c0re/src/dashboard.rs +++ b/hive-c0re/src/dashboard.rs @@ -76,13 +76,13 @@ pub async fn serve(port: u16, coord: Arc) -> Result<()> { .route("/op-send", post(post_op_send)) .route("/meta-update", post(post_meta_update)) .route("/api/schedules", get(api_schedules).post(post_schedule_new)) - .route( - "/api/schedules/{id}", - axum::routing::patch(patch_schedule), - ) + .route("/api/schedules/{id}", axum::routing::patch(patch_schedule)) .route("/api/schedules/{id}/cancel", post(post_schedule_cancel)) .route("/api/schedules/{id}/fire-now", post(post_schedule_fire_now)) - .route("/api/rebuild-queue/{id}/cancel", post(post_rebuild_queue_cancel)) + .route( + "/api/rebuild-queue/{id}/cancel", + post(post_rebuild_queue_cancel), + ) .route("/dashboard/stream", get(dashboard_stream)) .route("/dashboard/history", get(dashboard_history)) // Anything not matched by the dynamic routes above falls @@ -765,7 +765,14 @@ async fn dashboard_history(State(state): State) -> Response { let events: Vec = messages .into_iter() .map(|m| match m { - crate::broker::MessageEvent::Sent { id, from, to, body, at, in_reply_to } => { + crate::broker::MessageEvent::Sent { + id, + from, + to, + body, + at, + in_reply_to, + } => { let file_refs = scan_validated_paths(&body); crate::dashboard_events::DashboardEvent::Sent { seq: 0, @@ -778,7 +785,14 @@ async fn dashboard_history(State(state): State) -> Response { file_refs, } } - crate::broker::MessageEvent::Delivered { id, from, to, body, at, in_reply_to } => { + crate::broker::MessageEvent::Delivered { + id, + from, + to, + body, + at, + in_reply_to, + } => { let file_refs = scan_validated_paths(&body); crate::dashboard_events::DashboardEvent::Delivered { seq: 0, @@ -1155,8 +1169,8 @@ fn resolve_state_path( return Err(format!("path not in allow-list: {raw}")); }; reject_symlinks_below(std::path::Path::new(root), &mapped)?; - let canonical = std::fs::canonicalize(&mapped) - .map_err(|e| format!("{}: {e}", mapped.display()))?; + let canonical = + std::fs::canonicalize(&mapped).map_err(|e| format!("{}: {e}", mapped.display()))?; if !(canonical.starts_with(AGENTS_ROOT) || canonical.starts_with(SHARED_ROOT)) { return Err(format!( "resolved path escapes allow-list: {}", @@ -1174,8 +1188,8 @@ fn resolve_state_path( )); } } - let meta = std::fs::metadata(&canonical) - .map_err(|e| format!("stat {}: {e}", canonical.display()))?; + let meta = + std::fs::metadata(&canonical).map_err(|e| format!("stat {}: {e}", canonical.display()))?; if meta.is_file() { let mode = meta.permissions().mode(); if mode & 0o004 == 0 { @@ -1313,12 +1327,10 @@ pub(crate) async fn emit_tombstones_snapshot(coord: &Arc) { let containers = coord.containers_snapshot().await; let transient_snapshot = coord.transient_snapshot(); let tombstones = build_tombstone_views(coord, &containers, &transient_snapshot); - coord.emit_dashboard_event( - crate::dashboard_events::DashboardEvent::TombstonesChanged { - seq: coord.next_seq(), - tombstones, - }, - ); + coord.emit_dashboard_event(crate::dashboard_events::DashboardEvent::TombstonesChanged { + seq: coord.next_seq(), + tombstones, + }); } /// Snapshot meta/flake.lock's root inputs + emit @@ -1326,12 +1338,10 @@ pub(crate) async fn emit_tombstones_snapshot(coord: &Arc) { /// (`run_meta_update`, `auto_update::rebuild_agent`). pub(crate) fn emit_meta_inputs_snapshot(coord: &Coordinator) { let inputs = read_meta_inputs(); - coord.emit_dashboard_event( - crate::dashboard_events::DashboardEvent::MetaInputsChanged { - seq: coord.next_seq(), - inputs, - }, - ); + coord.emit_dashboard_event(crate::dashboard_events::DashboardEvent::MetaInputsChanged { + seq: coord.next_seq(), + inputs, + }); } /// Scan `body` for path-shaped tokens, validate each against the @@ -1381,9 +1391,7 @@ pub(crate) fn scan_validated_paths(body: &str) -> Vec { out } -async fn get_state_file( - axum::extract::Query(q): axum::extract::Query, -) -> Response { +async fn get_state_file(axum::extract::Query(q): axum::extract::Query) -> Response { const MAX_BYTES: usize = 1 << 20; // 1 MiB let (canonical, meta) = match resolve_state_path(&q.path) { Ok(pair) => pair, @@ -1415,11 +1423,18 @@ async fn get_state_file( return ([("content-type", ct)], bytes).into_response(); } let truncated = bytes.len() > MAX_BYTES; - let body_bytes = if truncated { &bytes[..MAX_BYTES] } else { &bytes[..] }; + let body_bytes = if truncated { + &bytes[..MAX_BYTES] + } else { + &bytes[..] + }; let mut body = String::from_utf8_lossy(body_bytes).into_owned(); if truncated { use std::fmt::Write as _; - let _ = write!(body, "\n\n--- truncated at {MAX_BYTES} of {size} bytes ---\n"); + let _ = write!( + body, + "\n\n--- truncated at {MAX_BYTES} of {size} bytes ---\n" + ); } ([("content-type", "text/plain; charset=utf-8")], body).into_response() } @@ -1492,7 +1507,7 @@ async fn post_schedule_new( description: payload.description, source: crate::scheduled_prompts::ScheduleSource::Operator, }; - match state.coord.scheduled_prompts.submit(new) { + match state.coord.scheduled_prompts.submit(&new) { Ok(id) => axum::Json(serde_json::json!({"id": id})).into_response(), Err(e) => error_response(&format!("schedule submit: {e:#}")), } @@ -1546,6 +1561,12 @@ struct CancelScheduleForm { } #[derive(serde::Deserialize, Default)] +#[allow( + clippy::option_option, + reason = "double-Option carries three-state PATCH semantics on the wire \ + (missing key = leave alone, JSON null = clear, value = set); \ + collapsing to a single Option would lose the 'clear' state" +)] struct EditScheduleForm { #[serde(default)] body: Option, @@ -1671,7 +1692,10 @@ async fn get_agent_links(AxumPath(name): AxumPath) -> Response { match client.get(&url).send().await { Ok(resp) if resp.status().is_success() => match resp.json::().await { Ok(body) => { - let links = body.get("links").cloned().unwrap_or_else(|| serde_json::json!([])); + let links = body + .get("links") + .cloned() + .unwrap_or_else(|| serde_json::json!([])); axum::Json(links).into_response() } Err(e) => { @@ -1852,7 +1876,10 @@ async fn post_op_send(State(state): State, Form(form): Form) -> Vec { .collect() } - /// Multi-file unified diff between the currently-deployed tree and /// the proposal for this approval. Runs against the applied repo /// since the canonical proposal commit lives there (manager-side @@ -2282,4 +2308,3 @@ async fn get_approval_diff( fn plain_text(body: String) -> Response { (StatusCode::OK, body).into_response() } - diff --git a/hive-c0re/src/dashboard_events.rs b/hive-c0re/src/dashboard_events.rs index 0ae794a8..83a3fcac 100644 --- a/hive-c0re/src/dashboard_events.rs +++ b/hive-c0re/src/dashboard_events.rs @@ -165,10 +165,7 @@ pub enum DashboardEvent { /// last one cached on the coordinator. Mutation sites (lifecycle /// endpoints, `actions::destroy` / approve, `crash_watch`'s poll loop) /// call the rescan after their work lands. - ContainerStateChanged { - seq: u64, - container: ContainerView, - }, + ContainerStateChanged { seq: u64, container: ContainerView }, /// A container that was in the previous snapshot is gone. Clients /// drop the row by name. Fired alongside any /// `nixos-container destroy` (operator-driven or otherwise) on the @@ -211,12 +208,9 @@ pub enum DashboardEvent { /// snapshot-shape rationale as `TombstonesChanged` / /// `MetaInputsChanged`: the list is small, snapshot semantics avoid /// the add/remove races a per-row event would have, and the - /// dashboard's grouping (parent_id) is most naturally re-derived + /// dashboard's grouping (`parent_id`) is most naturally re-derived /// from the full list. - RebuildQueueChanged { - seq: u64, - queue: Vec, - }, + RebuildQueueChanged { seq: u64, queue: Vec }, } impl DashboardEvent { @@ -259,13 +253,18 @@ mod tests { /// the `kind` JSON field matches `kind_tag()`. The exhaustive /// `match` in `kind_tag` already provides compile-time variant /// coverage — this test is the value-side guard against - /// typos in the snake_case strings vs serde's `rename_all` + /// typos in the `snake_case` strings vs serde's `rename_all` /// output. `ContainerStateChanged` is omitted from the sample /// list only because `ContainerView` has no `Default` impl and /// constructing one inline here is more boilerplate than the /// test is worth; the variant is still covered by the /// `kind_tag` match arm. #[test] + #[allow( + clippy::too_many_lines, + reason = "exhaustive coverage of every DashboardEvent variant — the \ + length is the point" + )] fn kind_tag_matches_serde_kind_field() { let samples: Vec = vec![ DashboardEvent::Sent { @@ -367,11 +366,7 @@ mod tests { .get("kind") .and_then(|k| k.as_str()) .expect("kind field present"); - assert_eq!( - ev.kind_tag(), - serde_kind, - "kind_tag() drift on {ev:?}", - ); + assert_eq!(ev.kind_tag(), serde_kind, "kind_tag() drift on {ev:?}",); } } } diff --git a/hive-c0re/src/manager_server.rs b/hive-c0re/src/manager_server.rs index b23cbf40..3eb20919 100644 --- a/hive-c0re/src/manager_server.rs +++ b/hive-c0re/src/manager_server.rs @@ -90,7 +90,11 @@ fn manager_recv_timeout(wait_seconds: Option) -> std::time::Duration { #[allow(clippy::too_many_lines)] async fn dispatch(req: &ManagerRequest, coord: &Arc) -> ManagerResponse { match req { - ManagerRequest::Send { to, body, in_reply_to } => { + ManagerRequest::Send { + to, + body, + in_reply_to, + } => { if let Err(message) = crate::limits::check_size("send", body) { return ManagerResponse::Err { message }; } @@ -195,7 +199,14 @@ async fn dispatch(req: &ManagerRequest, coord: &Arc) -> ManagerResp ) { Ok(id) => { tracing::info!(%id, %name, "init_config approval queued"); - coord.emit_approval_added(id, name, "init_config", None, None, description.clone()); + coord.emit_approval_added( + id, + name, + "init_config", + None, + None, + description.clone(), + ); ManagerResponse::Ok } Err(e) => ManagerResponse::Err { @@ -302,7 +313,7 @@ async fn dispatch(req: &ManagerRequest, coord: &Arc) -> ManagerResp Err(e) => { return ManagerResponse::Err { message: format!("queue update_meta_inputs approval: {e:#}"), - } + }; } }; tracing::info!(%id, %label, "update_meta_inputs approval queued"); @@ -317,7 +328,7 @@ async fn dispatch(req: &ManagerRequest, coord: &Arc) -> ManagerResp ManagerResponse::Ok } ManagerRequest::RequestSchedulePrompt(payload) => { - handle_request_schedule_prompt(coord, hive_sh4re::MANAGER_AGENT, payload).await + handle_request_schedule_prompt(coord, hive_sh4re::MANAGER_AGENT, payload) } ManagerRequest::CancelSchedule { id, targets } => { handle_cancel_schedule(coord, hive_sh4re::MANAGER_AGENT, *id, targets.as_deref()) @@ -482,12 +493,13 @@ async fn dispatch(req: &ManagerRequest, coord: &Arc) -> ManagerResp ManagerRequest::SetStatus { text } => { let path = Coordinator::agent_notes_dir(MANAGER_AGENT).join("hyperhive-status"); let result = if text.trim().is_empty() { - std::fs::remove_file(&path) - .or_else(|e| if e.kind() == std::io::ErrorKind::NotFound { + std::fs::remove_file(&path).or_else(|e| { + if e.kind() == std::io::ErrorKind::NotFound { Ok(()) } else { Err(e) - }) + } + }) } else { std::fs::write(&path, format!("{}\n", text.trim())) }; @@ -497,7 +509,9 @@ async fn dispatch(req: &ManagerRequest, coord: &Arc) -> ManagerResp tokio::spawn(async move { coord2.rescan_containers_and_emit().await }); ManagerResponse::Ok } - Err(e) => ManagerResponse::Err { message: format!("set_status write failed: {e}") }, + Err(e) => ManagerResponse::Err { + message: format!("set_status write failed: {e}"), + }, } } ManagerRequest::GetAgentMeta { name } => { @@ -508,7 +522,12 @@ async fn dispatch(req: &ManagerRequest, coord: &Arc) -> ManagerResp // tell (e.g. "iris is down" vs "iris has no status set"). let (status_text, status_set_at, running) = crate::container_view::read_agent_status_live(target).await; - let role = if target == MANAGER_AGENT { "manager" } else { "agent" }.to_owned(); + let role = if target == MANAGER_AGENT { + "manager" + } else { + "agent" + } + .to_owned(); ManagerResponse::AgentMeta { name: target.to_owned(), role, @@ -518,16 +537,12 @@ async fn dispatch(req: &ManagerRequest, coord: &Arc) -> ManagerResp status_set_at, } } - ManagerRequest::CancelLooseEnd { kind, id } => crate::questions::handle_cancel_loose_end( - coord, - MANAGER_AGENT, - *kind, - *id, - ) - .map_or_else( - |message| ManagerResponse::Err { message }, - |()| ManagerResponse::Ok, - ), + ManagerRequest::CancelLooseEnd { kind, id } => { + crate::questions::handle_cancel_loose_end(coord, MANAGER_AGENT, *kind, *id).map_or_else( + |message| ManagerResponse::Err { message }, + |()| ManagerResponse::Ok, + ) + } ManagerRequest::AckTurn => match coord.broker.ack_turn(MANAGER_AGENT) { Ok(_n) => ManagerResponse::Ok, Err(e) => ManagerResponse::Err { @@ -706,7 +721,7 @@ async fn submit_apply_commit( /// inputs (non-empty targets, non-empty body, sane interval) at /// submit time — the operator should never see a malformed schedule /// pending approval. -async fn handle_request_schedule_prompt( +fn handle_request_schedule_prompt( coord: &Arc, requester: &str, payload: &hive_sh4re::SchedulePromptPayload, @@ -731,7 +746,7 @@ async fn handle_request_schedule_prompt( Err(e) => { return ManagerResponse::Err { message: format!("encode SchedulePromptPayload: {e:#}"), - } + }; } }; let id = match coord.approvals.submit_kind( @@ -744,7 +759,7 @@ async fn handle_request_schedule_prompt( Err(e) => { return ManagerResponse::Err { message: format!("queue schedule_prompt approval: {e:#}"), - } + }; } }; tracing::info!( @@ -783,12 +798,12 @@ fn handle_cancel_schedule( Ok(None) => { return ManagerResponse::Err { message: format!("schedule {schedule_id} not found"), - } + }; } Err(e) => { return ManagerResponse::Err { message: format!("read schedule {schedule_id}: {e:#}"), - } + }; } }; if !cancel_authorized(requester, &schedule.owner) { @@ -830,12 +845,12 @@ async fn handle_fire_schedule_now( Ok(None) => { return ManagerResponse::Err { message: format!("schedule {schedule_id} not found"), - } + }; } Err(e) => { return ManagerResponse::Err { message: format!("read schedule {schedule_id}: {e:#}"), - } + }; } }; if !cancel_authorized(requester, &schedule.owner) { @@ -858,11 +873,16 @@ async fn handle_fire_schedule_now( /// ownership rules as `CancelSchedule` — the manager can edit /// schedules it owns + any owned by an agent in its subtree. /// Forwards the partial payload to -/// `ScheduledPrompts::update` which enforces the cancelled-row -/// + zero-interval validation. Returns `Ok` on a clean update; +/// `ScheduledPrompts::update` which enforces the cancelled-row / +/// zero-interval validation. Returns `Ok` on a clean update; /// `Err` with the underlying message on any auth / validation /// failure so the dashboard can surface it verbatim. #[allow(clippy::too_many_arguments)] +#[allow( + clippy::option_option, + reason = "double-Option carries three-state PATCH semantics: outer None = \ + leave alone, Some(None) = clear, Some(Some(v)) = set" +)] fn handle_edit_schedule( coord: &Arc, requester: &str, @@ -879,12 +899,12 @@ fn handle_edit_schedule( Ok(None) => { return ManagerResponse::Err { message: format!("schedule {schedule_id} not found"), - } + }; } Err(e) => { return ManagerResponse::Err { message: format!("read schedule {schedule_id}: {e:#}"), - } + }; } }; if !cancel_authorized(requester, &schedule.owner) { diff --git a/hive-c0re/src/meta.rs b/hive-c0re/src/meta.rs index dc3cd2f0..5b4f1167 100644 --- a/hive-c0re/src/meta.rs +++ b/hive-c0re/src/meta.rs @@ -73,7 +73,13 @@ pub async fn sync_agents( let dir = meta_dir(); std::fs::create_dir_all(&dir).with_context(|| format!("create {}", dir.display()))?; - let new_flake = render_flake(hyperhive_flake, dashboard_port, operator_pronouns, context_window_tokens, agents); + let new_flake = render_flake( + hyperhive_flake, + dashboard_port, + operator_pronouns, + context_window_tokens, + agents, + ); let flake_path = dir.join("flake.nix"); let on_disk = std::fs::read_to_string(&flake_path).unwrap_or_default(); let initial = !dir.join(".git").exists(); @@ -308,6 +314,11 @@ fn agent_canonical_inputs(name: &str) -> Vec<&'static str> { /// Inner render helper accepting a lookup fn so tests can stub the /// agent flake-lock introspection. +#[allow( + clippy::too_many_lines, + reason = "templated string-builder for the meta flake — the length is one \ + contiguous fmt block, splitting it would just hide the shape" +)] fn render_flake_with_lookup( hyperhive_flake: &str, dashboard_port: u16, @@ -404,16 +415,20 @@ where sorted_tokens.sort_by_key(|(k, _)| k.as_str()); for (key, val) in &sorted_tokens { let upper_key = key.to_ascii_uppercase(); - let _ = writeln!(out, " HIVE_CONTEXT_WINDOW_TOKENS_{upper_key} = \"{val}\";"); + let _ = writeln!( + out, + " HIVE_CONTEXT_WINDOW_TOKENS_{upper_key} = \"{val}\";" + ); } // Forge URL — injected when hive-c0re itself has HIVE_FORGE_URL set // (the NixOS module derives it from hyperhive.forge.{domain,httpPort}). // Agents use it in forge_notify to poll Forgejo for PR/review events. if let Ok(forge_url) = std::env::var("HIVE_FORGE_URL") - && !forge_url.is_empty() { - let escaped = forge_url.replace('\\', "\\\\").replace('"', "\\\""); - let _ = writeln!(out, " HIVE_FORGE_URL = \"{escaped}\";"); - } + && !forge_url.is_empty() + { + let escaped = forge_url.replace('\\', "\\\\").replace('"', "\\\""); + let _ = writeln!(out, " HIVE_FORGE_URL = \"{escaped}\";"); + } out.push_str( r#" HYPERHIVE_STATE_DIR = "/agents/${name}/state"; }; @@ -450,6 +465,82 @@ where out } +async fn git_is_clean(dir: &Path) -> Result { + let out = lifecycle::git_command() + .current_dir(dir) + .args(["status", "--porcelain"]) + .output() + .await + .with_context(|| format!("git status in {}", dir.display()))?; + Ok(out.stdout.iter().all(u8::is_ascii_whitespace)) +} + +async fn git(dir: &Path, args: &[&str]) -> Result<()> { + let out = lifecycle::git_command() + .current_dir(dir) + .args(args) + .output() + .await + .with_context(|| format!("git {} in {}", args.join(" "), dir.display()))?; + if !out.status.success() { + bail!( + "git {} failed ({}): {}", + args.join(" "), + out.status, + String::from_utf8_lossy(&out.stderr).trim() + ); + } + Ok(()) +} + +async fn git_commit(dir: &Path, message: &str) -> Result<()> { + git( + dir, + &[ + "-c", + &format!("user.name={GIT_NAME}"), + "-c", + &format!("user.email={GIT_EMAIL}"), + "commit", + "-m", + message, + ], + ) + .await?; + // Best-effort mirror to the bundled forge. No-op when the forge + // isn't seeded (no core token on disk); push failures log a warn + // but don't bubble up — a missing mirror shouldn't fail an + // otherwise successful deploy. + if let Err(e) = crate::forge::push_meta(dir).await { + tracing::warn!(error = ?e, "forge: meta push after commit failed (non-fatal)"); + } + Ok(()) +} + +async fn nix(dir: &Path, args: &[&str]) -> Result<()> { + // `--extra-experimental-features` belt-and-suspenders for hosts + // that haven't set this in nix.conf. The hyperhive module's + // deploy guide assumes flakes are already enabled, but the cost + // of being defensive is one extra argv each call. + let mut all = vec!["--extra-experimental-features", "nix-command flakes"]; + all.extend(args); + let out = Command::new("nix") + .current_dir(dir) + .args(&all) + .output() + .await + .with_context(|| format!("nix {} in {}", args.join(" "), dir.display()))?; + if !out.status.success() { + bail!( + "nix {} failed ({}): {}", + args.join(" "), + out.status, + String::from_utf8_lossy(&out.stderr).trim() + ); + } + Ok(()) +} + #[cfg(test)] mod tests { use super::*; @@ -557,79 +648,3 @@ mod tests { ); } } - -async fn git_is_clean(dir: &Path) -> Result { - let out = lifecycle::git_command() - .current_dir(dir) - .args(["status", "--porcelain"]) - .output() - .await - .with_context(|| format!("git status in {}", dir.display()))?; - Ok(out.stdout.iter().all(u8::is_ascii_whitespace)) -} - -async fn git(dir: &Path, args: &[&str]) -> Result<()> { - let out = lifecycle::git_command() - .current_dir(dir) - .args(args) - .output() - .await - .with_context(|| format!("git {} in {}", args.join(" "), dir.display()))?; - if !out.status.success() { - bail!( - "git {} failed ({}): {}", - args.join(" "), - out.status, - String::from_utf8_lossy(&out.stderr).trim() - ); - } - Ok(()) -} - -async fn git_commit(dir: &Path, message: &str) -> Result<()> { - git( - dir, - &[ - "-c", - &format!("user.name={GIT_NAME}"), - "-c", - &format!("user.email={GIT_EMAIL}"), - "commit", - "-m", - message, - ], - ) - .await?; - // Best-effort mirror to the bundled forge. No-op when the forge - // isn't seeded (no core token on disk); push failures log a warn - // but don't bubble up — a missing mirror shouldn't fail an - // otherwise successful deploy. - if let Err(e) = crate::forge::push_meta(dir).await { - tracing::warn!(error = ?e, "forge: meta push after commit failed (non-fatal)"); - } - Ok(()) -} - -async fn nix(dir: &Path, args: &[&str]) -> Result<()> { - // `--extra-experimental-features` belt-and-suspenders for hosts - // that haven't set this in nix.conf. The hyperhive module's - // deploy guide assumes flakes are already enabled, but the cost - // of being defensive is one extra argv each call. - let mut all = vec!["--extra-experimental-features", "nix-command flakes"]; - all.extend(args); - let out = Command::new("nix") - .current_dir(dir) - .args(&all) - .output() - .await - .with_context(|| format!("nix {} in {}", args.join(" "), dir.display()))?; - if !out.status.success() { - bail!( - "nix {} failed ({}): {}", - args.join(" "), - out.status, - String::from_utf8_lossy(&out.stderr).trim() - ); - } - Ok(()) -} diff --git a/hive-c0re/src/rebuild_queue.rs b/hive-c0re/src/rebuild_queue.rs index 35a35f72..e4010fe9 100644 --- a/hive-c0re/src/rebuild_queue.rs +++ b/hive-c0re/src/rebuild_queue.rs @@ -70,6 +70,7 @@ pub enum QueueKind { Spawn, /// Destroy with `--purge` (real fs work). Not yet routed here; the /// variant exists so the wire shape doesn't need to change later. + #[allow(dead_code, reason = "wire shape — routed by a future PR")] Destroy, } @@ -102,10 +103,11 @@ pub enum QueueSource { AutoUpdate, /// Crash recovery path (future use — currently no auto-rebuild on /// crash, but the variant exists for the imminent feature). + #[allow(dead_code, reason = "wire shape — used by a future feature")] CrashRecover, /// Operator approved a pending `Approval` row on the dashboard. /// `QueueEntry.approval_id` points back at the source row so the - /// worker can fetch the kind-specific payload (commit_ref, inputs, + /// worker can fetch the kind-specific payload (`commit_ref`, inputs, /// description) before dispatching. Approval, } @@ -137,7 +139,10 @@ pub enum QueueState { impl QueueState { pub fn is_terminal(self) -> bool { - matches!(self, QueueState::Done | QueueState::Failed | QueueState::Cancelled) + matches!( + self, + QueueState::Done | QueueState::Failed | QueueState::Cancelled + ) } } @@ -150,7 +155,7 @@ pub struct QueueEntry { /// so SSE upserts land in place rather than churning the list. pub id: u64, /// Target agent name, or the literal `"hyperhive"` for entries - /// (MetaUpdate) that affect the meta flake rather than a single + /// (`MetaUpdate`) that affect the meta flake rather than a single /// agent. pub agent: String, pub kind: QueueKind, @@ -183,8 +188,8 @@ pub struct QueueEntry { pub inputs: Vec, /// Source approval row id when this entry was created by an /// operator-approve POST (`source == Approval`). The worker uses - /// it to re-fetch the kind-specific payload (commit_ref / inputs / - /// description / fetched_sha) and to fire `ApprovalResolved` on + /// it to re-fetch the kind-specific payload (`commit_ref` / inputs / + /// description / `fetched_sha`) and to fire `ApprovalResolved` on /// completion. `None` for non-approval entries — preserved on /// the wire that way too. #[serde(default, skip_serializing_if = "Option::is_none")] @@ -312,7 +317,7 @@ impl RebuildQueue { // docstring + #365 for why). Approval-driven entries also // require the approval_id to match so two distinct approvals // for the same agent never collapse into one queue slot. - for entry in inner.entries.iter_mut() { + for entry in &mut inner.entries { if entry.state == QueueState::Queued && entry.kind == kind && entry.agent == agent @@ -320,7 +325,8 @@ impl RebuildQueue { && entry.approval_id == approval_id { if !entry.reason.contains(&reason) { - entry.reason.push_str(&format!("\nalso requested by: {reason}")); + use std::fmt::Write as _; + let _ = write!(entry.reason, "\nalso requested by: {reason}"); } return entry.id; } @@ -372,7 +378,10 @@ impl RebuildQueue { /// and leaving a stale "in flight" label after a terminal /// transition would mislead the dashboard render. pub fn finish(&self, id: u64, state: QueueState, error: Option) { - debug_assert!(state.is_terminal(), "finish() called with non-terminal {state:?}"); + debug_assert!( + state.is_terminal(), + "finish() called with non-terminal {state:?}" + ); let mut inner = self.inner.lock().expect("rebuild_queue mutex poisoned"); if let Some(entry) = inner.entries.iter_mut().find(|e| e.id == id) { entry.state = state; @@ -422,7 +431,7 @@ impl RebuildQueue { pub fn cancel_children(&self, parent: u64) -> usize { let mut inner = self.inner.lock().expect("rebuild_queue mutex poisoned"); let mut count = 0; - for entry in inner.entries.iter_mut() { + for entry in &mut inner.entries { if entry.parent_id == Some(parent) && entry.state == QueueState::Queued { entry.state = QueueState::Cancelled; entry.finished_at = Some(now_unix()); @@ -440,13 +449,13 @@ impl RebuildQueue { /// safely interrupted). Returns true when an entry was cancelled. pub fn cancel(&self, id: u64) -> bool { let mut inner = self.inner.lock().expect("rebuild_queue mutex poisoned"); - if let Some(entry) = inner.entries.iter_mut().find(|e| e.id == id) { - if entry.state == QueueState::Queued { - entry.state = QueueState::Cancelled; - entry.finished_at = Some(now_unix()); - Self::trim_history(&mut inner); - return true; - } + if let Some(entry) = inner.entries.iter_mut().find(|e| e.id == id) + && entry.state == QueueState::Queued + { + entry.state = QueueState::Cancelled; + entry.finished_at = Some(now_unix()); + Self::trim_history(&mut inner); + return true; } false } @@ -533,7 +542,7 @@ pub async fn run_worker(coord: std::sync::Arc) return; } } - _ = coord.rebuild_queue.notify.notified() => { + () = coord.rebuild_queue.notify.notified() => { // New entry — back to the drain loop. } } @@ -556,12 +565,14 @@ async fn dispatch( crate::actions::run_approval_apply_commit(coord, Some(entry.id), approval_id).await } (QueueKind::Rebuild, None) => { - let current_rev = crate::auto_update::current_flake_rev(&coord.hyperhive_flake) - .unwrap_or_default(); - crate::auto_update::rebuild_agent(coord, &entry.agent, ¤t_rev, Some(entry.id)).await + let current_rev = + crate::auto_update::current_flake_rev(&coord.hyperhive_flake).unwrap_or_default(); + crate::auto_update::rebuild_agent(coord, &entry.agent, ¤t_rev, Some(entry.id)) + .await } (QueueKind::MetaUpdate, Some(approval_id)) => { - crate::actions::run_approval_update_meta_inputs(coord, Some(entry.id), approval_id).await + crate::actions::run_approval_update_meta_inputs(coord, Some(entry.id), approval_id) + .await } (QueueKind::MetaUpdate, None) => run_meta_update(coord, entry).await, (QueueKind::Spawn, Some(approval_id)) => { @@ -601,7 +612,11 @@ async fn run_meta_update( ) -> anyhow::Result<()> { let _progress = coord.meta_update_guard(); let inputs = entry.inputs.clone(); - tracing::info!(?inputs, parent = entry.id, "rebuild_queue: meta-update starting"); + tracing::info!( + ?inputs, + parent = entry.id, + "rebuild_queue: meta-update starting" + ); coord.set_queue_step(Some(entry.id), "nix flake update"); let result = if inputs.is_empty() { crate::meta::lock_update(&[]).await @@ -633,7 +648,7 @@ async fn run_meta_update( /// Compute which agents a `nix flake update ` on the meta /// flake would affect. Used by callers that pre-enqueue cascade -/// `Rebuild` entries at MetaUpdate submission time (issue #347) so the +/// `Rebuild` entries at `MetaUpdate` submission time (issue #347) so the /// dashboard can render the dependent work alongside its parent before /// the lock bump actually runs. /// @@ -658,7 +673,8 @@ pub async fn meta_update_cascade_agents(inputs: &[String]) -> Vec { if c == crate::lifecycle::MANAGER_NAME { Some(crate::lifecycle::MANAGER_NAME.to_owned()) } else { - c.strip_prefix(crate::lifecycle::AGENT_PREFIX).map(str::to_owned) + c.strip_prefix(crate::lifecycle::AGENT_PREFIX) + .map(str::to_owned) } }) .collect() @@ -782,9 +798,11 @@ mod tests { // Both inputs lists are preserved. let inputs: Vec<&[String]> = snap.iter().map(|e| e.inputs.as_slice()).collect(); assert!(inputs.iter().any(|i| *i == ["nixpkgs"])); - assert!(inputs - .iter() - .any(|i| *i == ["agent-bitburner/bitburner-agent"])); + assert!( + inputs + .iter() + .any(|i| *i == ["agent-bitburner/bitburner-agent"]) + ); } #[test] diff --git a/hive-c0re/src/scheduled_prompts.rs b/hive-c0re/src/scheduled_prompts.rs index 8b958452..739f7751 100644 --- a/hive-c0re/src/scheduled_prompts.rs +++ b/hive-c0re/src/scheduled_prompts.rs @@ -164,6 +164,12 @@ pub struct NewSchedule { /// "this target is active again"; prior history was already visible /// at cancel time). #[derive(Debug, Clone, Default)] +#[allow( + clippy::option_option, + reason = "double-Option carries three-state PATCH semantics: outer None = \ + leave alone, Some(None) = clear, Some(Some(v)) = set. \ + collapsing to a single Option would lose the 'clear' state" +)] pub struct UpdateSchedule { pub body: Option, pub description: Option>, @@ -200,7 +206,7 @@ impl ScheduledPrompts { /// Insert a new schedule. Returns the new id. Empty `targets` is /// rejected — a schedule with no recipients would silently /// never fan out, masking caller bugs. - pub fn submit(&self, new: NewSchedule) -> Result { + pub fn submit(&self, new: &NewSchedule) -> Result { if new.targets.is_empty() { bail!("schedule must have at least one target"); } @@ -212,13 +218,13 @@ impl ScheduledPrompts { created_at_unix, source, description) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)", params![ - new.owner, - new.body, + &new.owner, + &new.body, new.interval_seconds.map(i64::try_from).and_then(Result::ok), new.first_fire_at_unix, now_unix(), new.source.to_db_string(), - new.description, + &new.description, ], )?; let id = tx.last_insert_rowid(); @@ -255,7 +261,7 @@ impl ScheduledPrompts { /// Every active (non-globally-cancelled) schedule in insert /// order. Used by the dashboard list view + the cancel-auth - /// check (the latter only needs the header but list() is the + /// check (the latter only needs the header but `list()` is the /// shared hot path). pub fn list(&self) -> Result> { let conn = self.conn.lock().unwrap(); @@ -322,7 +328,7 @@ impl ScheduledPrompts { /// Advance a recurring schedule's `next_fire_at` to the smallest /// multiple-of-interval > `from`. Returns the count of skipped /// cycles (≥ 0); the worker stamps that into the per-row - /// last_result so operators see "caught up from N missed". + /// `last_result` so operators see "caught up from N missed". /// /// For one-shots (`interval_seconds IS NULL`) this is a no-op /// at the SQL level; callers should `delete` them after fan-out @@ -488,10 +494,7 @@ impl ScheduledPrompts { /// id. pub fn delete(&self, id: i64) -> Result<()> { let conn = self.conn.lock().unwrap(); - conn.execute( - "DELETE FROM scheduled_prompts WHERE id = ?1", - params![id], - )?; + conn.execute("DELETE FROM scheduled_prompts WHERE id = ?1", params![id])?; Ok(()) } @@ -630,7 +633,7 @@ mod tests { } fn submit_one_shot(db: &ScheduledPrompts, fire_at: i64, targets: &[&str]) -> i64 { - db.submit(NewSchedule { + db.submit(&NewSchedule { owner: "operator".into(), targets: targets.iter().map(|t| (*t).to_owned()).collect(), body: "wake".into(), @@ -658,7 +661,7 @@ mod tests { fn submit_rejects_empty_targets() { let (_dir, db) = open(); let err = db - .submit(NewSchedule { + .submit(&NewSchedule { owner: "operator".into(), targets: Vec::new(), body: "wake".into(), @@ -688,7 +691,7 @@ mod tests { let (_dir, db) = open(); // Recurring every 60s, last fire at t=100. let id = db - .submit(NewSchedule { + .submit(&NewSchedule { owner: "operator".into(), targets: vec!["alice".into()], body: "wake".into(), @@ -710,7 +713,7 @@ mod tests { fn rearm_advances_one_step_when_caught_up() { let (_dir, db) = open(); let id = db - .submit(NewSchedule { + .submit(&NewSchedule { owner: "operator".into(), targets: vec!["alice".into()], body: "wake".into(), @@ -742,7 +745,8 @@ mod tests { fn cancel_targets_auto_cancels_parent_when_last_drops() { let (_dir, db) = open(); let id = submit_one_shot(&db, 100, &["alice", "bob"]); - db.cancel_targets(id, &["alice".to_owned()]).expect("cancel"); + db.cancel_targets(id, &["alice".to_owned()]) + .expect("cancel"); let s = db.get(id).expect("get").expect("present"); // Parent still active (bob remains). assert!(s.cancelled_at_unix.is_none()); @@ -775,7 +779,8 @@ mod tests { fn record_target_result_skips_cancelled_targets() { let (_dir, db) = open(); let id = submit_one_shot(&db, 100, &["alice", "bob"]); - db.cancel_targets(id, &["alice".to_owned()]).expect("cancel"); + db.cancel_targets(id, &["alice".to_owned()]) + .expect("cancel"); db.record_target_result(id, "alice", 200, "ok") .expect("record alice"); db.record_target_result(id, "bob", 200, "ok") @@ -794,7 +799,7 @@ mod tests { fn update_partial_only_touches_set_fields() { let (_dir, db) = open(); let id = db - .submit(NewSchedule { + .submit(&NewSchedule { owner: "operator".into(), targets: vec!["alice".into()], body: "old body".into(), @@ -824,7 +829,7 @@ mod tests { fn update_interval_toggle_recurring_to_one_shot() { let (_dir, db) = open(); let id = db - .submit(NewSchedule { + .submit(&NewSchedule { owner: "operator".into(), targets: vec!["alice".into()], body: "x".into(), @@ -899,7 +904,7 @@ mod tests { fn update_clears_description() { let (_dir, db) = open(); let id = db - .submit(NewSchedule { + .submit(&NewSchedule { owner: "operator".into(), targets: vec!["alice".into()], body: "x".into(), @@ -982,7 +987,8 @@ mod tests { let (_dir, db) = open(); let id = submit_one_shot(&db, 100, &["alice", "bob"]); // Record some history on alice, then cancel her. - db.record_target_result(id, "alice", 50, "ok").expect("record"); + db.record_target_result(id, "alice", 50, "ok") + .expect("record"); db.update( id, UpdateSchedule { @@ -1034,7 +1040,7 @@ mod tests { fn approval_source_round_trips() { let (_dir, db) = open(); let id = db - .submit(NewSchedule { + .submit(&NewSchedule { owner: "manager".into(), targets: vec!["alice".into()], body: "wake".into(), diff --git a/hive-c0re/src/scheduled_prompts_worker.rs b/hive-c0re/src/scheduled_prompts_worker.rs index 2489274d..7ab6a500 100644 --- a/hive-c0re/src/scheduled_prompts_worker.rs +++ b/hive-c0re/src/scheduled_prompts_worker.rs @@ -32,7 +32,7 @@ //! the broker send again, so transient errors self-heal. //! - **one-shots** delete unconditionally after their single //! fan-out pass; a broker failure on a one-shot is NOT -//! retried (the operator advisory + last_result are the only +//! retried (the operator advisory + `last_result` are the only //! audit trail). use std::sync::Arc; @@ -104,7 +104,7 @@ fn tick(coord: &Arc) { } /// Fan out one schedule's body to every active target. Records -/// per-target last_result; advances or reaps the parent row at +/// per-target `last_result`; advances or reaps the parent row at /// the end depending on whether `interval_seconds` is set. fn fire_schedule(coord: &Arc, schedule: &Schedule, now: i64) { let known: std::collections::HashSet = known_agents(coord); @@ -118,12 +118,11 @@ fn fire_schedule(coord: &Arc, schedule: &Schedule, now: i64) { // mirrors `to == operator` into its own pane. if target != hive_sh4re::OPERATOR_RECIPIENT && !known.contains(target) { let reason = format!("no such agent: {target}"); - if let Err(e) = coord.scheduled_prompts.record_target_result( - schedule.id, - target, - now, - &reason, - ) { + if let Err(e) = + coord + .scheduled_prompts + .record_target_result(schedule.id, target, now, &reason) + { tracing::warn!(error = ?e, schedule = schedule.id, %target, "record_target_result failed"); } notify_operator_missing_target(coord, schedule, target); @@ -292,7 +291,11 @@ pub async fn fire_now( if schedule.cancelled_at_unix.is_some() { anyhow::bail!("schedule {schedule_id} is already cancelled"); } - if !schedule.targets.iter().any(|t| t.cancelled_at_unix.is_none()) { + if !schedule + .targets + .iter() + .any(|t| t.cancelled_at_unix.is_none()) + { anyhow::bail!("schedule {schedule_id} has no active targets"); } let known = known_agents_async().await; @@ -309,12 +312,11 @@ pub async fn fire_now( let target = &target_row.target; if target != hive_sh4re::OPERATOR_RECIPIENT && !known.contains(target) { let reason = format!("manual fire: no such agent: {target}"); - if let Err(e) = coord.scheduled_prompts.record_target_result( - schedule_id, - target, - now, - &reason, - ) { + if let Err(e) = + coord + .scheduled_prompts + .record_target_result(schedule_id, target, now, &reason) + { tracing::warn!(error = ?e, schedule = schedule_id, %target, "record_target_result failed"); } notify_operator_missing_target(coord, &schedule, target); diff --git a/hive-c0re/src/topology.rs b/hive-c0re/src/topology.rs index 7830e9a9..71ef41e1 100644 --- a/hive-c0re/src/topology.rs +++ b/hive-c0re/src/topology.rs @@ -58,6 +58,11 @@ pub fn read() -> BTreeMap> { /// or absent from the file. Cheap convenience over `read()` for /// callers that want a single entry. #[must_use] +#[allow( + dead_code, + reason = "convenience API; callers go through `read()` today, kept for the \ + dashboard/manager-server surfaces landing in #361 follow-ups" +)] pub fn parent_of(name: &str) -> Option { read().get(name).cloned().flatten() } @@ -88,10 +93,10 @@ pub fn is_descendant_of(candidate: &str, ancestor: &str) -> bool { false } -/// Persist the topology map. Sorted JSON output (BTreeMap is sorted by +/// Persist the topology map. Sorted JSON output (`BTreeMap` is sorted by /// key) keeps git diffs minimal across re-writes. Best-effort — /// returns an `io::Error` so callers can decide whether a failure -/// should abort their op (sync_agents, RequestSetParent) or just log. +/// should abort their op (`sync_agents`, `RequestSetParent`) or just log. pub fn write(topology: &BTreeMap>) -> std::io::Result<()> { let path = topology_path(); if let Some(parent) = path.parent() { @@ -111,13 +116,21 @@ pub fn write(topology: &BTreeMap>) -> std::io::Result<()> /// entries — `sync_agents` only adds rows for newly-spawned agents /// against whatever the operator has configured. #[must_use] +#[allow( + dead_code, + reason = "kept for the dashboard / RequestSetParent write API landing in \ + #361 follow-ups; `sync_agents` does its own seeding today" +)] pub fn default_seed(agent_names: &[String]) -> BTreeMap> { let mut out = BTreeMap::new(); for name in agent_names { if name == crate::lifecycle::MANAGER_NAME { out.insert(name.clone(), None); } else { - out.insert(name.clone(), Some(crate::lifecycle::MANAGER_NAME.to_owned())); + out.insert( + name.clone(), + Some(crate::lifecycle::MANAGER_NAME.to_owned()), + ); } } out @@ -285,8 +298,8 @@ mod tests { #[test] fn apply_set_parent_refuses_manager_move() { - let err = - apply_set_parent(&topo_three_level(), crate::lifecycle::MANAGER_NAME, None).unwrap_err(); + let err = apply_set_parent(&topo_three_level(), crate::lifecycle::MANAGER_NAME, None) + .unwrap_err(); assert!(err.contains("manager"), "err = {err}"); } diff --git a/hive-sh4re/src/lib.rs b/hive-sh4re/src/lib.rs index cb8b8f6e..c69ad976 100644 --- a/hive-sh4re/src/lib.rs +++ b/hive-sh4re/src/lib.rs @@ -752,14 +752,10 @@ pub enum ManagerRequest { Status, /// Operator-injected message TO the manager (from the manager's own web /// UI). Same shape as `AgentRequest::OperatorMsg`. - OperatorMsg { - body: String, - }, + OperatorMsg { body: String }, /// Last `limit` messages addressed to the manager, newest-first. /// Non-mutating; mirror of `AgentRequest::Recent`. - Recent { - limit: u64, - }, + Recent { limit: u64 }, /// Initialise a brand-new agent's proposed config repo and queue an /// approval for the operator to review. On approval hive-c0re seeds /// `/agents//config/` with the default `agent.nix` template, @@ -776,23 +772,15 @@ pub enum ManagerRequest { description: Option, }, /// Stop a sub-agent (graceful). - Kill { - name: String, - }, + Kill { name: String }, /// Start a previously-stopped sub-agent container. - Start { - name: String, - }, + Start { name: String }, /// Restart a sub-agent container (stop + start). - Restart { - name: String, - }, + Restart { name: String }, /// Rebuild a sub-agent: re-applies the current hyperhive flake + /// agent.nix, restarts the container. No approval required — /// it's idempotent and the manager owns its own update cadence. - Update { - name: String, - }, + Update { name: String }, /// Submit a config commit for the user to approve. `commit_ref` must /// be a commit sha (7-40 hex chars, short or full) in the agent's /// proposed config repo — a branch or tag name is rejected so the