vinzenz/hyperhive
vinzenz/hyperhive
Watch
0
0
Fork
You've already forked hyperhive
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

clippy: fix lints that crane's cargoClippy properly enforces (#538)

Browse source 
The naersk → crane swap in the parent commit flips clippy from
silently passing to actually failing on `-D warnings` (naersk's
`mode = "clippy"` mangled the `--` separator so the deny never took
effect). This commit clears the surfaced lints so the workspace
builds clean under the new enforcement — every fix is mechanical and
preserves behaviour. Tests still pass (160 across the workspace).

Auto-fixes via `cargo clippy --fix`:
- `doc_markdown` (19 sites): bare identifiers in doc comments
  wrapped in backticks
- `format_in_format_args`, `explicit_into_iter_loop`,
  `redundant_closure_for_method_calls`, `useless_conversion`, and
  a few more — mechanical rewrites of the kind cargo can apply
  safely.

Hand-fixed:
- `match_same_arms` (forge_notify::is_atx_heading): two arms returning
  `true` collapsed into a single `matches!` pattern.
- `cast_sign_loss` + `format_push_string` (mcp.rs status formatter):
  guarded `i64 → u64` through `u64::try_from(…).unwrap_or(0)` (status
  timestamps are always positive in practice; clamp the skew edge to
  0) and swapped `out.push_str(&format!(…))` for `write!` into the
  buffer with an infallible-writer `let _ =`.
- `doc_lazy_continuation` in turn.rs + manager_server.rs + sh4re/lib.rs:
  doc paragraphs that the markdown parser was treating as list-item
  continuations got either a separating blank line or a `/`-for-`+`
  word swap so the parser stops seeing a list.
- `unused_async` (manager_server::handle_request_schedule_prompt):
  function has no `.await`; dropped the `async` and its `.await` call
  site.
- `needless_pass_by_value` (scheduled_prompts::submit): take
  `&NewSchedule` instead of moving the struct in; updated two prod
  callers and eight test sites to pass references.
- `type_complexity` (approvals::mark_cancelled): hoisted the
  7-tuple SELECT row shape into a `type CancelLookupRow = (…);` alias.

Allow-with-reason for intentional patterns:
- `option_option` (6 sites across dashboard / scheduled_prompts /
  manager_server): `Option<Option<T>>` carries three-state PATCH
  semantics (missing key = leave alone, `Some(None)` = clear,
  `Some(Some(v))` = set). Collapsing to `Option<T>` loses the
  "clear" state.
- `dead_code` (rebuild_queue::QueueKind::Destroy /
  QueueSource::CrashRecover; topology::parent_of / default_seed):
  wire-shape variants + API surfaces kept for the upcoming features
  (#361 follow-ups, future `Destroy` queue routing, crash-recovery
  path). Allowed at the variant / function level with the rationale
  in `reason = "…"`.
- `too_many_lines` on three specific call-sites: a 117-line
  exhaustive-variant test (dashboard_events::kind_tag_matches_…),
  the meta-flake string template renderer
  (meta::render_flake_with_lookup), and the notification poll loop
  (forge_notify::poll_once) — splitting any of them would just hide
  the contiguous shape they exist to keep visible.

`nix flake check` formatting target is still broken on main itself
(pre-existing nixfmt drift across ~28 files unrelated to this PR);
left alone here so the scope stays "crane port + lints the port
exposed" and the operator's review doesn't have to triage drive-by
nixfmt churn.
This commit is contained in:
iris 2026-05-29 01:42:06 +02:00 • committed by Mara
commit 9ed58ab96d
17 changed files with 643 additions and 463 deletions
Download patch file Download diff file Expand all files Collapse all files

44
hive-ag3nt/src/events.rs
View file

@ -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::<u64>()
&& v > 0 {
return v;
}
&& !suffix.is_empty()
&& m.contains(&suffix.to_ascii_lowercase())
&& let Ok(v) = val.trim().parse::<u64>()
&& 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::<u64>()
&& 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 });
}

155
hive-ag3nt/src/forge_notify.rs
View file

@ -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<serde_json::Value> {
async fn fetch_json(client: &reqwest::Client, url: &str, token: &str) -> Option<serde_json::Value> {
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 &notifications {
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)");
}
}
}
}
}

99
hive-ag3nt/src/mcp.rs
View file

@ -47,7 +47,7 @@ pub enum SocketReply {
Recent(Vec<hive_sh4re::InboxRow>),
Logs(String),
/// `list_schedules` result — used by the manager surface only;
/// AgentResponse has no equivalent variant.
/// `AgentResponse` has no equivalent variant.
Schedules(Vec<hive_sh4re::WireSchedule>),
LooseEnds(Vec<hive_sh4re::LooseEnd>),
PendingRemindersCount(u64),
@ -185,8 +185,7 @@ pub fn format_recv(resp: Result<SocketReply, anyhow::Error>) -> 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<SocketReply, anyhow::Error>) -> String {
match status_text {
None => out.push_str("\nstatus: <none>"),
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<SetStatusArgs>) -> 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 `<none>` when the \
target has never called `set_status` or has cleared it."
)]
async fn get_agent_meta(
&self,
Parameters(args): Parameters<GetAgentMetaArgs>,
) -> String {
async fn get_agent_meta(&self, Parameters(args): Parameters<GetAgentMetaArgs>) -> String {
let log = args.name.clone().unwrap_or_else(|| "<self>".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<u64>,
/// 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<FireScheduleNowArgs>,
) -> String {
async fn fire_schedule_now(&self, Parameters(args): Parameters<FireScheduleNowArgs>) -> 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<SetStatusArgs>) -> 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 `<none>` when the target has never called \
`set_status` or has cleared it."
)]
async fn get_agent_meta(
&self,
Parameters(args): Parameters<GetAgentMetaArgs>,
) -> String {
async fn get_agent_meta(&self, Parameters(args): Parameters<GetAgentMetaArgs>) -> String {
let log = args.name.clone().unwrap_or_else(|| "<self>".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()]

96
hive-ag3nt/src/turn.rs
View file

@ -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::<serde_json::Value>(&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::<serde_json::Value>(&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}"),
});
}
}
});

26
hive-c0re/src/actions.rs
View file

@ -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<Coordinator>, 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:#}")),

24
hive-c0re/src/approvals.rs
View file

@ -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<Approval> {
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<String>,
Option<String>,
)> = tx
);
let mut conn = self.conn.lock().unwrap();
let tx = conn.transaction()?;
let row: Option<CancelLookupRow> = 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<Item = rusqlite::Result<Approval>>,
) -> Vec<Approval> {
fn collect_lenient(rows: impl Iterator<Item = rusqlite::Result<Approval>>) -> Vec<Approval> {
rows.filter_map(|r| match r {
Ok(a) => Some(a),
Err(e) => {
@ -467,7 +470,12 @@ mod tests {
// status + a "cancelled by <who>" 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);

7
hive-c0re/src/auto_update.rs
View file

@ -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<Coordinator>) -> 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())

81
hive-c0re/src/crash_watch.rs
View file

@ -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<String>,
current: &HashSet<String>,
sub_agents: &[String],
prev_sub_agents: &HashSet<String>,
) {
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<String>,
current: &HashSet<String>,
sub_agents: &[String],
prev_sub_agents: &HashSet<String>,
) {
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(),
});
}
}

91
hive-c0re/src/dashboard.rs
View file

@ -76,13 +76,13 @@ pub async fn serve(port: u16, coord: Arc<Coordinator>) -> 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<AppState>) -> Response {
let events: Vec<crate::dashboard_events::DashboardEvent> = 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<AppState>) -> 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<Coordinator>) {
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<Coordinator>) {
/// (`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<String> {
out
}
async fn get_state_file(
axum::extract::Query(q): axum::extract::Query<StateFileQuery>,
) -> Response {
async fn get_state_file(axum::extract::Query(q): axum::extract::Query<StateFileQuery>) -> 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<String>,
@ -1671,7 +1692,10 @@ async fn get_agent_links(AxumPath(name): AxumPath<String>) -> Response {
match client.get(&url).send().await {
Ok(resp) if resp.status().is_success() => match resp.json::<serde_json::Value>().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<AppState>, Form(form): Form<OpSendForm
.coord
.broadcast_send(hive_sh4re::OPERATOR_RECIPIENT, &body);
if !errors.is_empty() {
return error_response(&format!("op-send broadcast partial fail: {}", errors.join("; ")));
return error_response(&format!(
"op-send broadcast partial fail: {}",
errors.join("; ")
));
}
} else if let Err(e) = state.coord.broker.send(&hive_sh4re::Message {
from: hive_sh4re::OPERATOR_RECIPIENT.to_owned(),
@ -2151,7 +2178,6 @@ fn gc_orphans(coord: &Coordinator, approvals: Vec<Approval>) -> Vec<Approval> {
.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()
}

25
hive-c0re/src/dashboard_events.rs
View file

@ -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<QueueEntry>,
},
RebuildQueueChanged { seq: u64, queue: Vec<QueueEntry> },
}
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<DashboardEvent> = 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:?}",);
}
}
}

80
hive-c0re/src/manager_server.rs
View file

@ -90,7 +90,11 @@ fn manager_recv_timeout(wait_seconds: Option<u64>) -> std::time::Duration {
#[allow(clippy::too_many_lines)]
async fn dispatch(req: &ManagerRequest, coord: &Arc<Coordinator>) -> 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<Coordinator>) -> 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<Coordinator>) -> 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<Coordinator>) -> 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<Coordinator>) -> 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<Coordinator>) -> 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<Coordinator>) -> 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<Coordinator>) -> 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<Coordinator>,
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<Coordinator>,
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) {

179
hive-c0re/src/meta.rs
View file

@ -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<F>(
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<bool> {
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<bool> {
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(())
}

72
hive-c0re/src/rebuild_queue.rs
View file

@ -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<String>,
/// 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<String>) {
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<crate::coordinator::Coordinator>)
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, &current_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, &current_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 <inputs>` 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<String> {
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]

48
hive-c0re/src/scheduled_prompts.rs
View file

@ -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<String>,
pub description: Option<Option<String>>,
@ -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<i64> {
pub fn submit(&self, new: &NewSchedule) -> Result<i64> {
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<Vec<Schedule>> {
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(),

32
hive-c0re/src/scheduled_prompts_worker.rs
View file

@ -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<Coordinator>) {
}
/// 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<Coordinator>, schedule: &Schedule, now: i64) {
let known: std::collections::HashSet<String> = known_agents(coord);
@ -118,12 +118,11 @@ fn fire_schedule(coord: &Arc<Coordinator>, 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);

23
hive-c0re/src/topology.rs
View file

@ -58,6 +58,11 @@ pub fn read() -> BTreeMap<String, Option<String>> {
/// 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<String> {
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<String, Option<String>>) -> std::io::Result<()> {
let path = topology_path();
if let Some(parent) = path.parent() {
@ -111,13 +116,21 @@ pub fn write(topology: &BTreeMap<String, Option<String>>) -> 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<String, Option<String>> {
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}");
}

24
hive-sh4re/src/lib.rs
View file

@ -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/<name>/config/` with the default `agent.nix` template,
@ -776,23 +772,15 @@ pub enum ManagerRequest {
description: Option<String>,
},
/// 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