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

Compare commits

base: vinzenz:30d82148e01c2936e98ea202bfd62c9cd20ff70e
Branches Tags
vinzenz:main
...
compare: vinzenz:908cadb151e20561ed3fefbfdad808942958a3d8
Branches Tags
vinzenz:main
Author SHA1 Message Date
damocles
908cadb151 docs: add missing #[must_use], # Errors, # Panics across public api 2026-05-22 19:41:27 +02:00
damocles
748536203b refactor: split long functions per review feedback; remove all #[allow] attributes 2026-05-22 19:24:44 +02:00
damocles
bbe2112dc9 fix: use try_from for i64/u64 casts; split format_notification into helpers 2026-05-22 19:14:22 +02:00
damocles
484cea62c7 fix #265: resolve all remaining clippy warnings (cast, too_many_lines, doc nits) 2026-05-22 19:11:40 +02:00
20 changed files with 833 additions and 643 deletions
Download patch file Download diff file Expand all files Collapse all files

201
hive-ag3nt/src/bin/hive-ag3nt.rs
View file

@ -149,11 +149,11 @@ async fn main() -> Result<()> {
}
}
#[allow(clippy::too_many_arguments, clippy::similar_names)]
#[allow(clippy::too_many_arguments)]
async fn serve(
socket: &Path,
interval: Duration,
state: Arc<Mutex<LoginState>>,
_login_state: Arc<Mutex<LoginState>>,
bus: Bus,
stats: Option<TurnStats>,
files: &turn::TurnFiles,
@ -161,25 +161,12 @@ async fn serve(
label: &str,
) -> Result<()> {
tracing::info!(socket = %socket.display(), "hive-ag3nt serve");
let _ = state; // reserved for future state transitions (turn-loop -> needs-login)
// Boot-time recovery: ask the broker to resurface anything we
// popped in a previous harness session but never acked
// (crashed mid-turn / OOM / container restart). The broker
// resets `delivered_at = NULL` on those rows and remembers
// their ids so the next `Recv` tags them `redelivered: true`;
// we then prepend a "may already be handled" hint to the wake
// prompt. Single shot before entering the serve loop; idempotent
// when there's nothing inflight.
requeue_inflight(socket).await;
loop {
let recv: Result<AgentResponse> =
// Explicit long-poll: the new agent_server semantics treat
// `None` as "peek, don't wait", which would tight-loop on
// sleep(interval). The harness wants to park until a
// message arrives, so opt into the full 180s cap.
// `max: None` (= 1) — the serve loop drives one turn per
// wake; claude itself calls recv(max: N) in-turn to drain
// a burst when the wake prompt mentions pending.
// Explicit long-poll: park until a message arrives (180s cap).
// `max: None` (= 1) — one turn per wake; claude calls
// recv(max: N) in-turn to drain bursts.
client::request(
socket,
&AgentRequest::Recv {
@ -191,93 +178,7 @@ async fn serve(
match recv {
Ok(AgentResponse::Messages { messages }) if !messages.is_empty() => {
let first = messages.into_iter().next().expect("checked non-empty");
let from = first.from;
let body = first.body;
let redelivered = first.redelivered;
tracing::info!(%from, %body, %redelivered, "inbox");
let unread = inbox_unread(socket).await;
bus.emit(LiveEvent::TurnStart {
from: from.clone(),
body: body.clone(),
unread,
});
bus.set_state(TurnState::Thinking);
let started_at = serve_common::now_unix();
let started_instant = std::time::Instant::now();
let model_at_start = bus.model();
let prompt = serve_common::format_wake_prompt(&from, &body, unread, redelivered);
let outcome = {
let _guard = turn_lock.lock().await;
turn::drive_turn(&prompt, files, &bus).await
};
turn::emit_turn_end(&bus, &outcome);
bus.set_state(TurnState::Idle);
// Ack only on a clean turn-end. `Failed` leaves every
// message popped during the turn in the unacked list;
// next harness boot's `RequeueInflight` will reset
// `delivered_at = NULL` and tag them `redelivered`.
// `PromptTooLong` is absorbed inside `drive_turn` via
// compaction so it shouldn't reach here, but if it
// does we also skip the ack (safer to redeliver than
// to lose the message).
if matches!(outcome, turn::TurnOutcome::Ok | turn::TurnOutcome::Compacted) {
ack_turn(socket).await;
}
// Rate-limited: park until the quota resets, then requeue
// the unacked message so it resurfaces in the same session.
if matches!(outcome, turn::TurnOutcome::RateLimited) {
let secs = turn::rate_limit_sleep_secs();
bus.emit_status("rate_limited");
bus.emit(LiveEvent::Note {
text: format!(
"API rate-limited — sleeping {secs}s before retry"
),
});
tracing::warn!(sleep_secs = secs, "rate-limited; parking");
tokio::time::sleep(Duration::from_secs(secs)).await;
requeue_inflight(socket).await;
bus.emit_status("online");
}
// Failures are unhandled by definition — PromptTooLong is
// absorbed inside drive_turn via compaction, so anything
// that reaches Failed here is a real crash. Notify the
// manager so it can investigate / restart / page the
// operator; best-effort, swallow the send error.
if let turn::TurnOutcome::Failed(e) = &outcome {
notify_manager_of_failure(socket, label, e).await;
}
if let Some(s) = &stats {
let ended_at = serve_common::now_unix();
let duration_ms =
i64::try_from(started_instant.elapsed().as_millis()).unwrap_or(i64::MAX);
let (open_threads, open_reminders) = fetch_agent_post_turn_counts(socket).await;
let row = serve_common::build_row(
started_at,
ended_at,
duration_ms,
model_at_start,
from.clone(),
&outcome,
&bus,
open_threads,
open_reminders,
);
s.record(&row);
}
// After turn completes, log whether messages arrived during
// the turn — the outer loop will iterate back to recv() on
// its own (the Empty-arm sleep only fires when recv
// actually returned Empty), so no explicit continue needed.
let pending = inbox_unread(socket).await;
if pending > 0 {
tracing::info!(%pending, "pending messages after turn; fetching next");
}
// `request_next_turn` MCP tool: agent wrote a sentinel
// requesting an immediate self-continuation turn. Clear
// the file and inject a synthetic wake so the outer loop
// fires a bare turn even if the inbox is empty.
check_and_inject_continue(socket, label).await;
handle_agent_turn(socket, &bus, stats.as_ref(), files, &turn_lock, label, first).await;
}
Ok(AgentResponse::Messages { .. }) => {
// Idle: empty list = nothing pending. Brief sleep
@ -307,14 +208,88 @@ async fn serve(
}
}
/// Per-turn user prompt. The role/tools/etc. is in the system prompt
/// (`prompts/agent.md` → `claude --system-prompt-file`); this is just the
/// wake signal claude reacts to. `unread` is the count of *other*
/// messages in the inbox right after this one was popped.
/// `redelivered` flags messages that were popped in a prior harness
/// session, never acked, and resurfaced after a restart — a banner
/// at the top of the wake prompt warns that any side-effects of
/// previous handling may already have happened.
/// Drive one turn for a received agent-inbox message.
async fn handle_agent_turn(
socket: &Path,
bus: &Bus,
stats: Option<&TurnStats>,
files: &turn::TurnFiles,
turn_lock: &TurnLock,
label: &str,
first: hive_sh4re::DeliveredMessage,
) {
let from = first.from;
let body = first.body;
let redelivered = first.redelivered;
tracing::info!(%from, %body, %redelivered, "inbox");
let unread = inbox_unread(socket).await;
bus.emit(LiveEvent::TurnStart { from: from.clone(), body: body.clone(), unread });
bus.set_state(TurnState::Thinking);
let started_at = serve_common::now_unix();
let started_instant = std::time::Instant::now();
let model_at_start = bus.model();
let prompt = serve_common::format_wake_prompt(&from, &body, unread, redelivered);
let outcome = {
let _guard = turn_lock.lock().await;
turn::drive_turn(&prompt, files, bus).await
};
turn::emit_turn_end(bus, &outcome);
bus.set_state(TurnState::Idle);
// Ack only on a clean turn-end. `Failed` leaves every message popped
// during the turn in the unacked list; next harness boot requeues them.
if matches!(outcome, turn::TurnOutcome::Ok | turn::TurnOutcome::Compacted) {
ack_turn(socket).await;
}
if matches!(outcome, turn::TurnOutcome::RateLimited) {
let secs = turn::rate_limit_sleep_secs();
bus.emit_status("rate_limited");
bus.emit(LiveEvent::Note {
text: format!("API rate-limited — sleeping {secs}s before retry"),
});
tracing::warn!(sleep_secs = secs, "rate-limited; parking");
tokio::time::sleep(Duration::from_secs(secs)).await;
requeue_inflight(socket).await;
bus.emit_status("online");
}
// Real crash: PromptTooLong is absorbed by compaction inside drive_turn.
if let turn::TurnOutcome::Failed(e) = &outcome {
notify_manager_of_failure(socket, label, e).await;
}
if let Some(stats) = stats {
let ended_at = serve_common::now_unix();
let duration_ms =
i64::try_from(started_instant.elapsed().as_millis()).unwrap_or(i64::MAX);
let (open_threads, open_reminders) = fetch_agent_post_turn_counts(socket).await;
let row = serve_common::build_row(
started_at,
ended_at,
duration_ms,
model_at_start,
from.clone(),
&outcome,
bus,
open_threads,
open_reminders,
);
stats.record(&row);
}
let pending = inbox_unread(socket).await;
if pending > 0 {
tracing::info!(%pending, "pending messages after turn; fetching next");
}
// `request_next_turn` MCP tool: agent wrote a sentinel requesting
// an immediate self-continuation. Clear and inject synthetic wake.
check_and_inject_continue(socket, label).await;
}
// Per-turn user prompt: the role/tools/etc. is in the system prompt
// (`prompts/agent.md` → `claude --system-prompt-file`); this is just the
// wake signal claude reacts to. `unread` is the count of *other*
// messages in the inbox right after this one was popped.
// `redelivered` flags messages that were popped in a prior harness
// session, never acked, and resurfaced after a restart — a banner
// at the top of the wake prompt warns that any side-effects of
// previous handling may already have happened.
/// Best-effort: tell the broker every message we popped during the
/// turn is now fully handled (turn-end-OK). Swallows transport

166
hive-ag3nt/src/bin/hive-m1nd.rs
View file

@ -144,93 +144,7 @@ async fn serve(
match recv {
Ok(ManagerResponse::Messages { messages }) if !messages.is_empty() => {
let first = messages.into_iter().next().expect("checked non-empty");
let from = first.from;
let body = first.body;
let redelivered = first.redelivered;
if from == SYSTEM_SENDER {
// Helper events (ApprovalResolved / Spawned / Rebuilt /
// Killed / Destroyed) — these are FYI for the manager;
// we surface them in the live view and forward them as
// a normal claude turn so the manager can react (e.g.
// greet a newly-spawned agent, retry a failed rebuild).
let parsed = serde_json::from_str::<HelperEvent>(&body).ok();
if let Some(event) = parsed {
tracing::info!(?event, "helper event");
} else {
tracing::info!(%from, %body, "system message");
}
bus.emit(LiveEvent::Note {
text: format!("[system] {body}"),
});
// Fall through: drive a turn with the event in the wake
// prompt body so claude sees it. Sender stays "system"
// so the wake prompt can label it as such.
}
tracing::info!(%from, %body, %redelivered, "manager inbox");
let unread = inbox_unread(socket).await;
bus.emit(LiveEvent::TurnStart {
from: from.clone(),
body: body.clone(),
unread,
});
let prompt = serve_common::format_wake_prompt(&from, &body, unread, redelivered);
bus.set_state(TurnState::Thinking);
let started_at = serve_common::now_unix();
let started_instant = std::time::Instant::now();
let model_at_start = bus.model();
let outcome = {
let _guard = turn_lock.lock().await;
turn::drive_turn(&prompt, files, &bus).await
};
turn::emit_turn_end(&bus, &outcome);
bus.set_state(TurnState::Idle);
// Ack only on a clean turn-end; Failed / RateLimited leave
// the popped ids in-flight for the next boot's requeue.
// Mirrors hive-ag3nt; see that loop for full rationale.
if matches!(outcome, turn::TurnOutcome::Ok | turn::TurnOutcome::Compacted) {
ack_turn(socket).await;
}
// Rate-limited: park until the quota resets, then requeue
// the unacked message so it resurfaces in the same session.
if matches!(outcome, turn::TurnOutcome::RateLimited) {
let secs = turn::rate_limit_sleep_secs();
bus.emit_status("rate_limited");
bus.emit(LiveEvent::Note {
text: format!(
"API rate-limited — sleeping {secs}s before retry"
),
});
tracing::warn!(sleep_secs = secs, "rate-limited; parking");
tokio::time::sleep(Duration::from_secs(secs)).await;
requeue_inflight(socket).await;
bus.emit_status("online");
}
if let Some(s) = &stats {
let ended_at = serve_common::now_unix();
let duration_ms =
i64::try_from(started_instant.elapsed().as_millis()).unwrap_or(i64::MAX);
let (open_threads, open_reminders) =
fetch_manager_post_turn_counts(socket).await;
let row = serve_common::build_row(
started_at,
ended_at,
duration_ms,
model_at_start,
from.clone(),
&outcome,
&bus,
open_threads,
open_reminders,
);
s.record(&row);
}
// Check for messages that arrived during the turn so we
// surface "draining" in the logs. The loop will already
// re-iterate from here — no explicit continue needed.
let pending = inbox_unread(socket).await;
if pending > 0 {
tracing::info!(%pending, "pending messages after turn; fetching next");
}
handle_manager_turn(socket, &bus, stats.as_ref(), files, &turn_lock, first).await;
}
Ok(ManagerResponse::Messages { .. }) => {
// Idle: empty list = nothing pending. Brief sleep
@ -260,6 +174,84 @@ async fn serve(
}
}
/// Drive one turn for a received manager-inbox message. Called from the
/// serve loop for the non-empty-messages arm to keep that loop readable.
async fn handle_manager_turn(
socket: &Path,
bus: &Bus,
stats: Option<&TurnStats>,
files: &turn::TurnFiles,
turn_lock: &TurnLock,
first: hive_sh4re::DeliveredMessage,
) {
let from = first.from;
let body = first.body;
let redelivered = first.redelivered;
if from == SYSTEM_SENDER {
// Helper events (ApprovalResolved / Spawned / Rebuilt /
// Killed / Destroyed) — surface in the live view and drive a
// normal turn so the manager can react.
let parsed = serde_json::from_str::<HelperEvent>(&body).ok();
if let Some(event) = parsed {
tracing::info!(?event, "helper event");
} else {
tracing::info!(%from, %body, "system message");
}
bus.emit(LiveEvent::Note { text: format!("[system] {body}") });
}
tracing::info!(%from, %body, %redelivered, "manager inbox");
let unread = inbox_unread(socket).await;
bus.emit(LiveEvent::TurnStart { from: from.clone(), body: body.clone(), unread });
let prompt = serve_common::format_wake_prompt(&from, &body, unread, redelivered);
bus.set_state(TurnState::Thinking);
let started_at = serve_common::now_unix();
let started_instant = std::time::Instant::now();
let model_at_start = bus.model();
let outcome = {
let _guard = turn_lock.lock().await;
turn::drive_turn(&prompt, files, bus).await
};
turn::emit_turn_end(bus, &outcome);
bus.set_state(TurnState::Idle);
// Ack only on a clean turn-end; Failed / RateLimited leave the
// popped ids in-flight for the next boot's requeue.
if matches!(outcome, turn::TurnOutcome::Ok | turn::TurnOutcome::Compacted) {
ack_turn(socket).await;
}
if matches!(outcome, turn::TurnOutcome::RateLimited) {
let secs = turn::rate_limit_sleep_secs();
bus.emit_status("rate_limited");
bus.emit(LiveEvent::Note {
text: format!("API rate-limited — sleeping {secs}s before retry"),
});
tracing::warn!(sleep_secs = secs, "rate-limited; parking");
tokio::time::sleep(Duration::from_secs(secs)).await;
requeue_inflight(socket).await;
bus.emit_status("online");
}
if let Some(stats) = stats {
let ended_at = serve_common::now_unix();
let duration_ms =
i64::try_from(started_instant.elapsed().as_millis()).unwrap_or(i64::MAX);
let (open_threads, open_reminders) = fetch_manager_post_turn_counts(socket).await;
let row = serve_common::build_row(
started_at,
ended_at,
duration_ms,
model_at_start,
from.clone(),
&outcome,
bus,
open_threads,
open_reminders,
);
stats.record(&row);
}
let pending = inbox_unread(socket).await;
if pending > 0 {
tracing::info!(%pending, "pending messages after turn; fetching next");
}
}
/// Best-effort: tell the broker every message popped during the turn
/// is now handled. Mirror of `hive-ag3nt::ack_turn` on the manager

15
hive-ag3nt/src/client.rs
View file

@ -19,6 +19,11 @@ const RETRY_BACKOFFS_MS: &[u64] = &[2_000, 4_000, 8_000, 16_000, 30_000];
/// the retry count. Use this from non-tool callers (the harness serve
/// loop, web UI, CLI subcommands) where we just want the socket-restart
/// resilience without surfacing the bookkeeping.
///
/// # Errors
///
/// Returns an error if the socket is unreachable after all retries, or if
/// serialization / deserialization of the request or response fails.
pub async fn request<Req, Resp>(socket: &Path, req: &Req) -> Result<Resp>
where
Req: Serialize + ?Sized,
@ -33,6 +38,16 @@ where
/// retries happened — that way claude knows the prior socket flake
/// wasn't a content error and shouldn't trigger an LLM-level retry of
/// its own.
///
/// # Errors
///
/// Returns an error if all retries are exhausted, or on a fatal protocol
/// error (serialization / deserialization failure).
///
/// # Panics
///
/// Panics if `RETRY_BACKOFFS_MS.len()` does not fit in a `u32`, which
/// cannot happen with the current compile-time constant.
pub async fn request_retried<Req, Resp>(socket: &Path, req: &Req) -> Result<(Resp, u32)>
where
Req: Serialize + ?Sized,

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

@ -222,6 +222,7 @@ pub struct TokenUsage {
impl TokenUsage {
/// Total context consumed this turn (input + cache reads + cache writes).
#[must_use]
pub fn context_tokens(&self) -> u64 {
self.input_tokens + self.cache_read_input_tokens + self.cache_creation_input_tokens
}
@ -230,6 +231,7 @@ impl TokenUsage {
/// **cumulative** sum across every inference in the turn — useful as a
/// cost signal, but NOT the current context size (a tool-heavy turn
/// sums per-call cached prompts and easily exceeds the model window).
#[must_use]
pub fn from_stream_event(v: &serde_json::Value) -> Option<Self> {
if v.get("type").and_then(|t| t.as_str()) != Some("result") {
return None;
@ -241,6 +243,7 @@ impl TokenUsage {
/// `.message.usage` block. Each turn fires one of these for every
/// model call; tracking the LAST one over the turn gives the actual
/// conversation context size — the number to watch for compaction.
#[must_use]
pub fn from_assistant_event(v: &serde_json::Value) -> Option<Self> {
if v.get("type").and_then(|t| t.as_str()) != Some("assistant") {
return None;
@ -443,12 +446,17 @@ impl Bus {
/// Take + clear the one-shot. Returns true iff the caller should
/// run claude without `--continue` for this turn.
#[must_use]
pub fn take_skip_continue(&self) -> bool {
self.skip_continue_once.swap(false, Ordering::SeqCst)
}
/// Currently-selected claude model name. Read on every turn so a
/// `/model <name>` flip takes effect on the next turn.
///
/// # Panics
///
/// Panics if the internal lock is poisoned.
#[must_use]
pub fn model(&self) -> String {
self.model.lock().unwrap().clone()
@ -459,6 +467,10 @@ impl Bus {
/// state dir (`hyperhive-model`) so the override survives harness
/// restart and container rebuild (gone on `--purge`, matching
/// every other piece of agent state).
///
/// # Panics
///
/// Panics if the internal lock is poisoned.
pub fn set_model(&self, name: impl Into<String>) {
let value: String = name.into();
self.model.lock().unwrap().clone_from(&value);
@ -472,6 +484,10 @@ impl Bus {
/// emitting a SSE event. Used by the bin entrypoints to backfill
/// from the most recent `turn_stats` row so the per-agent web UI's
/// ctx + cost badges paint real numbers on cold load.
///
/// # Panics
///
/// Panics if an internal lock is poisoned.
pub fn seed_usage(&self, ctx: Option<TokenUsage>, cost: Option<TokenUsage>) {
if ctx.is_some() {
*self.last_ctx_usage.lock().unwrap() = ctx;
@ -485,6 +501,10 @@ impl Bus {
/// usage (current context size); `cost` is the cumulative across
/// every inference in the turn (cost signal). One SSE event fires
/// per turn carrying both.
///
/// # Panics
///
/// Panics if an internal lock is poisoned.
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);
@ -503,6 +523,10 @@ impl Bus {
/// per-turn counter for each one we find. Called by the stdout
/// pump on every parsed line. Cheap when the line isn't an
/// assistant message — the field-check short-circuits.
///
/// # Panics
///
/// Panics if the internal lock is poisoned.
pub fn observe_stream(&self, v: &serde_json::Value) {
if v.get("type").and_then(|t| t.as_str()) != Some("assistant") {
return;
@ -531,6 +555,10 @@ impl Bus {
/// Snapshot + clear the per-turn tool-call counter. The harness
/// calls this between turns to fold the breakdown into a
/// `turn_stats` row, then start the next turn with an empty map.
///
/// # Panics
///
/// Panics if the internal lock is poisoned.
#[must_use]
pub fn take_tool_calls(&self) -> std::collections::HashMap<String, u64> {
std::mem::take(&mut *self.tool_calls.lock().unwrap())
@ -538,6 +566,10 @@ impl Bus {
/// Last context-size snapshot (last inference of the most recent
/// turn), or `None` if no turn has completed yet.
///
/// # Panics
///
/// Panics if the internal lock is poisoned.
#[must_use]
pub fn last_ctx_usage(&self) -> Option<TokenUsage> {
*self.last_ctx_usage.lock().unwrap()
@ -545,6 +577,10 @@ impl Bus {
/// Last cumulative cost snapshot (sum across the most recent turn's
/// inferences), or `None` if no turn has completed yet.
///
/// # Panics
///
/// Panics if the internal lock is poisoned.
#[must_use]
pub fn last_cost_usage(&self) -> Option<TokenUsage> {
*self.last_cost_usage.lock().unwrap()
@ -552,6 +588,10 @@ impl Bus {
/// Update the harness's authoritative turn-loop state. Records
/// the transition time so `state_snapshot` can return a since-age.
///
/// # Panics
///
/// Panics if the internal lock is poisoned.
pub fn set_state(&self, next: TurnState) {
let since;
{
@ -598,6 +638,10 @@ impl Bus {
}
/// Current state + since-when (unix seconds). Snapshot copy, no lock held.
///
/// # Panics
///
/// Panics if the internal lock is poisoned.
#[must_use]
pub fn state_snapshot(&self) -> (TurnState, i64) {
*self.state.lock().unwrap()
@ -617,6 +661,7 @@ impl Bus {
let _ = self.tx.send(envelope);
}
#[must_use]
pub fn subscribe(&self) -> broadcast::Receiver<BusEvent> {
self.tx.subscribe()
}

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

@ -22,6 +22,7 @@
//! are silently marked read — the agent already knows it opened them.
//! - Comment notifications where the comment author matches this agent's own
//! forge login are silently marked read.
//!
//! Own login is fetched once at startup via `GET /user` and cached for the
//! lifetime of the polling loop.
//!
@ -32,6 +33,7 @@
//! generic `[comment on PR #N repo]` so agents can action it immediately.
use std::collections::HashSet;
use std::fmt::Write as _;
use std::path::{Path, PathBuf};
use std::time::Duration;
@ -253,157 +255,185 @@ async fn format_notification(
};
let is_pr = matches!(notif_type, "Pull Request" | "Pull");
// Build assignee + reviewer suffix appended to all notification kinds.
let meta_suffix = {
let assignees: Vec<&str> = subject
.as_ref()
.and_then(|s| s["assignees"].as_array())
.map(|arr| arr.iter().filter_map(|a| a["login"].as_str()).collect())
.unwrap_or_default();
let assignee_line = if assignees.is_empty() {
"assignee: unassigned".to_owned()
} else {
format!("assignee: {}", assignees.join(", "))
};
// For PRs, include requested_reviewers when present.
let reviewer_line = if is_pr {
let reviewers: Vec<&str> = subject
.as_ref()
.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(", ")))
}
} else {
None
};
match reviewer_line {
Some(r) => format!("\n{assignee_line}\n{r}"),
None => format!("\n{assignee_line}"),
}
};
let meta_suffix = build_meta_suffix(subject.as_ref(), is_pr);
// Determine whether this notification was triggered by a comment/review or
// 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, subject, is_pr };
if has_comment {
// Notification triggered by a new comment or review submission.
let payload = fetch_json(client, comment_api_url, token).await;
let actor_login = payload
.as_ref()
.and_then(|c| c["user"]["login"].as_str())
.unwrap_or("");
// Self-notification filter (#230): skip if we authored the comment/review.
if !own_login.is_empty() && actor_login == own_login {
debug!(%own_login, "forge_notify: skipping self-authored comment/review");
return None;
}
let body_text = payload
.as_ref()
.and_then(|c| c["body"].as_str())
.unwrap_or("")
.trim();
// PR review detection (#231): Forgejo review objects carry a `state` field
// with values like "APPROVED" / "REQUEST_CHANGES" / "COMMENT". Regular
// issue/PR comments have no such field. Format reviews distinctly so the
// agent knows the review outcome immediately without reading the body.
let review_state = payload
.as_ref()
.and_then(|c| c["state"].as_str())
.and_then(review_state_label);
let url = if comment_html_url.is_empty() { html_url } else { comment_html_url };
let author = if actor_login.is_empty() { "?" } else { actor_login };
if let Some(review_label) = review_state {
// Review submission on a PR.
let kind = format!("PR {review_label}{num}{repo}");
let mut out = format!("[{kind}] {title}\nurl: {url}");
if body_text.is_empty() {
out.push_str(&format!("\n\nreviewer: {author}"));
} else {
out.push_str(&format!("\n\n{author}: {}", truncate(body_text, BODY_TRUNCATE)));
}
out.push_str(&meta_suffix);
Some(out)
} else {
// Regular comment.
let kind = format!("comment on {}{num}{repo}", notif_type_label(notif_type));
let mut out = format!(
"[{kind}] {title}\nurl: {url}\n\n{author}: {}",
truncate(body_text, BODY_TRUNCATE)
);
if out.ends_with('\n') {
out.pop();
}
out.push_str(&meta_suffix);
Some(out)
}
format_comment_notification(client, token, &meta, comment_api_url, comment_html_url, own_login).await
} else {
// Notification triggered by creation or state change of the subject.
//
// Classification uses notif["subject"]["state"] directly — Forgejo
// returns "open" / "closed" / "merged" here. We do NOT rely on
// fetching the PR/issue detail for `merged`:
// - `subject.url` points to the *issues* endpoint, which returns
// `pull_request.merged`, not top-level `merged`.
// - Forgejo API type is "Pull" / "Issue", never "Pull Request".
let notif_state = notif["subject"]["state"].as_str().unwrap_or("");
let reason = notif["reason"].as_str().unwrap_or("");
format_state_change_notification(notif, &meta, own_login)
}
}
// Self-notification filter (#230): skip new items we authored ourselves.
// `reason == "author"` combined with open state means we just opened the
// issue/PR. We do NOT filter merged/closed state changes — those are
// triggered by someone else and we want them.
let is_new = notif_state == "open" || notif_state.is_empty();
if is_new && reason == "author" && !own_login.is_empty() {
debug!(%own_login, "forge_notify: skipping self-authored new item");
return None;
}
/// Shared notification metadata extracted from the raw Forgejo JSON.
struct NotifMeta<'a> {
title: &'a str,
notif_type: &'a str,
html_url: &'a str,
num: String,
repo: String,
meta_suffix: String,
/// Fetched subject detail (issue/PR JSON); used for review-request detection.
subject: Option<serde_json::Value>,
is_pr: bool,
}
let label = notif_type_label(notif_type);
let kind = match notif_state {
"merged" => format!("{label} merged{num}{repo}"),
"closed" => format!("{label} closed{num}{repo}"),
"open" | "" => format!("new {label}{num}{repo}"),
other => format!("{label}{num}{repo}: {other}"),
};
/// Build the `\nassignee: ...` (and optionally `\nreviewer: ...`) suffix
/// appended to all notification kinds.
fn build_meta_suffix(subject: Option<&serde_json::Value>, is_pr: bool) -> String {
let assignees: Vec<&str> = subject
.and_then(|s| s["assignees"].as_array())
.map(|arr| arr.iter().filter_map(|a| a["login"].as_str()).collect())
.unwrap_or_default();
let assignee_line = if assignees.is_empty() {
"assignee: unassigned".to_owned()
} else {
format!("assignee: {}", assignees.join(", "))
};
// For PRs, include requested_reviewers when present.
let reviewer_line = if is_pr {
let reviewers: Vec<&str> = subject
.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(", "))) }
} else {
None
};
match reviewer_line {
Some(r) => format!("\n{assignee_line}\n{r}"),
None => format!("\n{assignee_line}"),
}
}
// Review-request detection (#253): Forgejo does not always set
// reason == "review_requested" (observed reason is null). Check
// requested_reviewers instead, which is reliable. If own_login is
// in the list, this is a review request -- override the kind.
// `subject` and `is_pr` are already fetched unconditionally above (#256).
let is_review_request = is_new
&& is_pr
&& !own_login.is_empty()
&& subject
.as_ref()
.and_then(|s| s["requested_reviewers"].as_array())
.map(|arr| arr.iter().any(|r| r["login"].as_str() == Some(own_login)))
.unwrap_or(false);
/// Format a notification triggered by a new comment or review submission.
async fn format_comment_notification(
client: &reqwest::Client,
token: &str,
meta: &NotifMeta<'_>,
comment_api_url: &str,
comment_html_url: &str,
own_login: &str,
) -> Option<String> {
let payload = fetch_json(client, comment_api_url, token).await;
let kind = if is_review_request {
format!("review requested{num}{repo}")
let actor_login = payload
.as_ref()
.and_then(|c| c["user"]["login"].as_str())
.unwrap_or("");
// Self-notification filter (#230): skip if we authored the comment/review.
if !own_login.is_empty() && actor_login == own_login {
debug!(%own_login, "forge_notify: skipping self-authored comment/review");
return None;
}
let body_text = payload
.as_ref()
.and_then(|c| c["body"].as_str())
.unwrap_or("")
.trim();
// PR review detection (#231): Forgejo review objects carry a `state` field
// with values like "APPROVED" / "REQUEST_CHANGES" / "COMMENT". Regular
// issue/PR comments have no such field. Format reviews distinctly so the
// agent knows the review outcome immediately without reading the body.
let review_state = payload
.as_ref()
.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;
if let Some(review_label) = review_state {
// Review submission on a PR.
let kind = format!("PR {review_label}{num}{repo}");
let mut out = format!("[{kind}] {title}\nurl: {url}");
if body_text.is_empty() {
write!(out, "\n\nreviewer: {author}").ok();
} else {
kind
};
let mut out = format!("[{kind}] {title}\nurl: {html_url}");
out.push_str(&meta_suffix);
write!(out, "\n\n{author}: {}", truncate(body_text, BODY_TRUNCATE)).ok();
}
out.push_str(meta_suffix);
Some(out)
} else {
// Regular comment.
let kind = format!("comment on {}{num}{repo}", notif_type_label(notif_type));
let mut out = format!(
"[{kind}] {title}\nurl: {url}\n\n{author}: {}",
truncate(body_text, BODY_TRUNCATE)
);
if out.ends_with('\n') {
out.pop();
}
out.push_str(meta_suffix);
Some(out)
}
}
/// Format a notification triggered by creation or state change of the subject.
fn format_state_change_notification(
notif: &serde_json::Value,
meta: &NotifMeta<'_>,
own_login: &str,
) -> Option<String> {
// Classification uses notif["subject"]["state"] directly — Forgejo
// returns "open" / "closed" / "merged" here. We do NOT rely on
// fetching the PR/issue detail for `merged`:
// - `subject.url` points to the *issues* endpoint, which returns
// `pull_request.merged`, not top-level `merged`.
// - Forgejo API type is "Pull" / "Issue", never "Pull Request".
let notif_state = notif["subject"]["state"].as_str().unwrap_or("");
let reason = notif["reason"].as_str().unwrap_or("");
// Self-notification filter (#230): skip new items we authored ourselves.
// `reason == "author"` combined with open state means we just opened the
// issue/PR. We do NOT filter merged/closed state changes — those are
// triggered by someone else and we want them.
let is_new = notif_state == "open" || notif_state.is_empty();
if is_new && reason == "author" && !own_login.is_empty() {
debug!(%own_login, "forge_notify: skipping self-authored new item");
return None;
}
let NotifMeta { title, notif_type, html_url, num, repo, meta_suffix, subject, is_pr } = meta;
let label = notif_type_label(notif_type);
let kind = match notif_state {
"merged" => format!("{label} merged{num}{repo}"),
"closed" => format!("{label} closed{num}{repo}"),
"open" | "" => format!("new {label}{num}{repo}"),
other => format!("{label}{num}{repo}: {other}"),
};
// Review-request detection (#253): Forgejo does not always set
// reason == "review_requested" (observed as null). Check
// requested_reviewers instead, which is reliable. If own_login is
// in the list, override the kind.
// subject and is_pr are already fetched unconditionally above (#256).
let is_review_request = is_new
&& *is_pr
&& !own_login.is_empty()
&& subject
.as_ref()
.and_then(|s| s["requested_reviewers"].as_array())
.is_some_and(|arr| arr.iter().any(|r| r["login"].as_str() == Some(own_login)));
let kind = if is_review_request {
format!("review requested{num}{repo}")
} else {
kind
};
let mut out = format!("[{kind}] {title}\nurl: {html_url}");
out.push_str(meta_suffix);
Some(out)
}
#[allow(clippy::too_many_arguments)]
async fn poll_once(
client: &reqwest::Client,
@ -449,15 +479,12 @@ async fn poll_once(
debug!(count = notifications.len(), "forge_notify: delivering notifications");
for notif in &notifications {
let id = match notif["id"].as_u64() {
Some(n) => n,
None => continue,
};
let Some(id) = notif["id"].as_u64() else { continue };
let body_opt = format_notification(client, token, notif, own_login).await;
// None means self-echo — mark read silently, no delivery.
let body = if let Some(b) = body_opt { b } else {
let Some(body) = body_opt else {
mark_read(client, forge_url, token, id).await;
continue;
};

38
hive-ag3nt/src/login_session.rs
View file

@ -48,6 +48,11 @@ impl LoginSession {
/// `HYPERHIVE_LOGIN_CMD` (single string, shell-split into argv); by
/// default we run `claude auth login`. Failing to spawn returns an error
/// before any state is registered.
///
/// # Errors
///
/// Returns an error if spawning the login command fails, or if the child's
/// stdio handles cannot be acquired.
pub fn start() -> Result<Self> {
let (cmd, args) = resolve_command();
tracing::info!(%cmd, ?args, "spawning login session");
@ -82,6 +87,11 @@ impl LoginSession {
/// Write `code` (plus a newline) to the child's stdin. Returns an error
/// if the stdin has already been closed (e.g. after the child exited or
/// after a prior submission consumed it).
///
/// # Errors
///
/// Returns an error if the login stdin is already closed, or if writing
/// to or flushing the stdin pipe fails.
pub async fn submit_code(&self, code: &str) -> Result<()> {
let mut guard = self.stdin.lock().await;
let stdin = guard.as_mut().context("login stdin already closed")?;
@ -100,18 +110,34 @@ impl LoginSession {
let _ = self.stdin.lock().await.take();
}
/// # Panics
///
/// Panics if the internal lock is poisoned.
#[must_use]
pub fn output(&self) -> String {
self.state.lock().unwrap().output.clone()
}
/// # Panics
///
/// Panics if the internal lock is poisoned.
#[must_use]
pub fn url(&self) -> Option<String> {
self.state.lock().unwrap().url.clone()
}
/// # Panics
///
/// Panics if the internal lock is poisoned.
#[must_use]
pub fn finished(&self) -> bool {
self.state.lock().unwrap().finished
}
/// # Panics
///
/// Panics if the internal lock is poisoned.
#[must_use]
pub fn exit_note(&self) -> Option<String> {
self.state.lock().unwrap().exit_note.clone()
}
@ -119,6 +145,10 @@ impl LoginSession {
/// Best-effort: poll the child once and update `finished`/`exit_note`.
/// Called by the web UI on each render so the state stays fresh without
/// running a dedicated reaper task.
///
/// # Panics
///
/// Panics if an internal lock is poisoned.
pub fn poll(&self) {
let mut child = self.child.lock().unwrap();
match child.try_wait() {
@ -137,6 +167,10 @@ impl LoginSession {
}
/// Kill the child if it's still running. Idempotent.
///
/// # Panics
///
/// Panics if the internal lock is poisoned.
pub fn kill(&self) {
if let Err(e) = self.child.lock().unwrap().start_kill() {
tracing::warn!(error = ?e, "kill login child");
@ -217,6 +251,10 @@ fn extract_url(line: &str) -> Option<String> {
/// Helper used by the web UI to gate "is there a session running right now"
/// without holding both this module's mutex and the `AppState`'s at once.
///
/// # Panics
///
/// Panics if the internal lock is poisoned.
pub fn drop_if_finished(slot: &Mutex<Option<Arc<LoginSession>>>) {
let mut guard = slot.lock().unwrap();
if let Some(s) = guard.as_ref() {

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

@ -114,6 +114,7 @@ impl From<hive_sh4re::ManagerResponse> for SocketReply {
/// Format helper for "send-like" tools (anything that expects an `Ok`).
/// `tool` and `ok_msg` only appear in the result string; they don't change
/// behavior.
#[must_use]
pub fn format_ack(resp: Result<SocketReply, anyhow::Error>, tool: &str, ok_msg: String) -> String {
match resp {
Ok(SocketReply::Ok) => ok_msg,
@ -131,6 +132,7 @@ pub fn format_ack(resp: Result<SocketReply, anyhow::Error>, tool: &str, ok_msg:
/// and `---` separators between bodies so the model can tell where
/// one ends and the next begins; per-message redelivery banners
/// included.
#[must_use]
pub fn format_recv(resp: Result<SocketReply, anyhow::Error>) -> String {
use std::fmt::Write as _;
let messages = match resp {
@ -171,6 +173,7 @@ pub const REDELIVERY_HINT: &str =
/// of pending approvals + questions + reminders. Empty list collapses
/// to a clear marker so claude doesn't go hunting for a payload that
/// isn't there.
#[must_use]
pub fn format_loose_ends(resp: Result<SocketReply, anyhow::Error>) -> String {
use std::fmt::Write as _;
let loose_ends = match resp {
@ -259,6 +262,7 @@ fn loose_end_kind_label(kind: hive_sh4re::CancelLooseEndKind) -> &'static str {
/// Format helper for `whoami`: renders the identity block as a short
/// human-readable string. Skips fields that are `None` so the output
/// doesn't carry dead placeholders.
#[must_use]
pub fn format_whoami(resp: Result<SocketReply, anyhow::Error>) -> String {
match resp {
Ok(SocketReply::Whoami {
@ -294,6 +298,7 @@ where
/// from "c0re flickered and the harness rode it out" — without the
/// hint, a tool result that took 30s to come back looks identical to a
/// content failure and the model would burn a turn retrying it.
#[must_use]
pub fn annotate_retries(mut s: String, retries: u32) -> String {
if retries > 0 {
use std::fmt::Write as _;
@ -660,6 +665,11 @@ impl AgentServer {
impl ServerHandler for AgentServer {}
/// Run the agent MCP server over stdio. Returns when the client disconnects.
///
/// # Errors
///
/// Returns an error if the MCP server fails to initialize or the transport
/// encounters a fatal error.
pub async fn serve_agent_stdio(socket: PathBuf) -> Result<()> {
let server = AgentServer::new(socket);
let service = server.serve(stdio()).await?;
@ -668,6 +678,11 @@ pub async fn serve_agent_stdio(socket: PathBuf) -> Result<()> {
}
/// Run the manager MCP server over stdio. Same idea, different tool surface.
///
/// # Errors
///
/// Returns an error if the MCP server fails to initialize or the transport
/// encounters a fatal error.
pub async fn serve_manager_stdio(socket: PathBuf) -> Result<()> {
let server = ManagerServer::new(socket);
let service = server.serve(stdio()).await?;

3
hive-ag3nt/src/serve_common.rs
View file

@ -12,6 +12,7 @@ use crate::turn_stats::TurnStatRow;
/// system prompt; this is just the wake signal body. `unread` is the inbox
/// depth after this message was popped. `redelivered` prepends a "may already
/// be handled" banner.
#[must_use]
pub fn format_wake_prompt(from: &str, body: &str, unread: u64, redelivered: bool) -> String {
let banner = if redelivered { REDELIVERY_HINT } else { "" };
let pending = if unread == 0 {
@ -26,6 +27,7 @@ pub fn format_wake_prompt(from: &str, body: &str, unread: u64, redelivered: bool
}
/// Current time as a Unix timestamp (seconds). Returns 0 on any error.
#[must_use]
pub fn now_unix() -> i64 {
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
@ -37,6 +39,7 @@ pub fn now_unix() -> i64 {
/// Assemble a `TurnStatRow` from the harness's per-turn state. Used by both
/// the agent and manager serve loops — the shape is identical, only the
/// post-turn count fetch helpers differ (and those stay in each binary).
#[must_use]
#[allow(clippy::too_many_arguments)]
pub fn build_row(
started_at: i64,

2
hive-ag3nt/src/stats.rs
View file

@ -29,6 +29,7 @@ pub enum Window {
}
impl Window {
#[must_use]
pub fn parse(s: &str) -> Self {
match s {
"1h" => Self::Hour,
@ -51,6 +52,7 @@ impl Window {
}
}
#[must_use]
pub fn span_secs(self) -> i64 {
match self {
Self::Hour => 3600,

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

@ -97,6 +97,10 @@ pub struct TurnFiles {
impl TurnFiles {
/// Write all three files into the per-agent runtime dir alongside
/// `socket`. Idempotent — overwrites whatever was there.
///
/// # Errors
///
/// Returns an error if any of the config files cannot be written to disk.
pub async fn prepare(socket: &Path, label: &str, flavor: mcp::Flavor) -> Result<Self> {
Ok(Self {
mcp_config: write_mcp_config(socket).await?,
@ -112,6 +116,10 @@ impl TurnFiles {
/// as `--socket <path>`); `binary_subcommand` is e.g. `"mcp"` for sub-agents
/// or `"mcp"` for the manager (both binaries name their MCP subcommand the
/// same — the differentiator is which binary `/proc/self/exe` resolves to).
///
/// # Errors
///
/// Returns an error if the config file cannot be written.
pub async fn write_mcp_config(socket: &Path) -> Result<PathBuf> {
let parent = socket.parent().unwrap_or_else(|| Path::new("/run/hive"));
tokio::fs::create_dir_all(parent).await.ok();
@ -128,6 +136,10 @@ pub async fn write_mcp_config(socket: &Path) -> Result<PathBuf> {
/// Drop the static `--settings` JSON next to the MCP config so we can
/// pass a path (`--settings <file>`) instead of an ever-growing inline
/// blob — the CLI argv has a finite length budget.
///
/// # Errors
///
/// Returns an error if the settings file cannot be written.
pub async fn write_settings(socket: &Path) -> Result<PathBuf> {
let parent = socket.parent().unwrap_or_else(|| Path::new("/run/hive"));
tokio::fs::create_dir_all(parent).await.ok();
@ -142,6 +154,10 @@ pub async fn write_settings(socket: &Path) -> Result<PathBuf> {
/// `--system-prompt-file`, replacing claude's default system prompt with
/// the role + tools instructions. Per-turn prompts become much smaller
/// (just the wake message body).
///
/// # Errors
///
/// Returns an error if the system prompt file cannot be written.
pub async fn write_system_prompt(
socket: &Path,
label: &str,
@ -198,6 +214,7 @@ pub fn rate_limit_sleep_secs() -> u64 {
/// 1. `HIVE_AUTO_RESET_WATERMARK_TOKENS` env var (explicit override).
/// 2. 50% of the model's context window (derived from `bus.model()` +
/// `events::context_window_tokens`).
///
/// `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")
@ -223,6 +240,7 @@ fn cache_ttl_secs() -> u64 {
/// 1. `HIVE_COMPACT_WATERMARK_TOKENS` env var (explicit override).
/// 2. 75% of the model's context window (derived from `bus.model()` +
/// `events::context_window_tokens`).
///
/// `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")
@ -397,6 +415,10 @@ pub fn emit_turn_end(bus: &Bus, outcome: &TurnOutcome) {
/// Block until the bound `~/.claude/` dir contains a session, polling
/// `claude_dir` on a `poll_ms` interval (min 2s). Flips `state` to
/// `Online` when login lands; caller resumes its serve loop.
///
/// # Panics
///
/// Panics if the internal login-state lock is poisoned.
pub async fn wait_for_login(
claude_dir: &Path,
state: Arc<Mutex<LoginState>>,
@ -440,6 +462,11 @@ pub async fn run_turn(prompt: &str, files: &TurnFiles, bus: &Bus) -> TurnOutcome
/// surface, same system prompt, same allowed-tools — so the post-
/// compact state matches a normal turn's. Only the prompt over stdin
/// differs (`/compact` vs the wake-up payload).
///
/// # Errors
///
/// Returns an error if the `claude --print /compact` invocation fails
/// (non-zero exit or I/O error).
pub async fn compact_session(files: &TurnFiles, bus: &Bus) -> Result<()> {
bus.emit(LiveEvent::Note {
text: "context overflow — running /compact on the persistent session".into(),

7
hive-ag3nt/src/turn_stats.rs
View file

@ -148,6 +148,10 @@ impl TurnStats {
/// Insert a row. Best-effort — logs + swallows errors so a sqlite
/// hiccup (locked db, full disk) doesn't crash the harness.
///
/// # Panics
///
/// Panics if the internal lock is poisoned.
pub fn record(&self, row: &TurnStatRow) {
let conn = self.inner.lock().unwrap();
let res = conn.execute(
@ -209,6 +213,9 @@ impl TurnStats {
/// have last-inference zeros — those rows yield `ctx = None` so the
/// badge stays empty until the next real turn rather than showing a
/// misleading 0.
/// # Panics
///
/// Panics if the internal lock is poisoned.
#[must_use]
pub fn last_usage(
&self,

6
hive-ag3nt/src/web_ui.rs
View file

@ -75,6 +75,9 @@ impl AppState {
/// `post_compact`) the allowed-tools surface claude sees.
pub type Flavor = mcp::Flavor;
/// # Errors
///
/// Returns an error if the TCP listener cannot bind to the given port.
pub async fn serve(
label: String,
port: u16,
@ -335,7 +338,8 @@ async fn api_stats(
// Pass the window span to the reminder-stats RPC so the broker
// filters its counts to the same time range as the chart data.
let window_secs = window.span_secs();
snapshot.reminder_stats = fetch_reminder_stats(&state.socket, state.flavor(), window_secs as u64).await;
let window_secs_u = u64::try_from(window_secs).unwrap_or(0);
snapshot.reminder_stats = fetch_reminder_stats(&state.socket, state.flavor(), window_secs_u).await;
axum::Json(snapshot)
}

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

@ -32,142 +32,164 @@ pub async fn approve(coord: Arc<Coordinator>, id: i64) -> Result<()> {
%approval.commit_ref,
"approval: running action",
);
let agent_dir = coord.ensure_runtime(&approval.agent)?;
let proposed_dir = Coordinator::agent_proposed_dir(&approval.agent);
let applied_dir = Coordinator::agent_applied_dir(&approval.agent);
let claude_dir = Coordinator::agent_claude_dir(&approval.agent);
let notes_dir = Coordinator::agent_notes_dir(&approval.agent);
match approval.kind {
ApprovalKind::ApplyCommit => {
let (result, terminal_tag, is_first_spawn) = run_apply_commit(
&coord,
&approval,
&agent_dir,
&applied_dir,
&claude_dir,
&notes_dir,
)
.await;
// Mirror the applied repo's new tag/branch state (approved/
// building/deployed-or-failed + main) to the forge.
if let Err(e) = crate::forge::push_config(&approval.agent).await {
tracing::warn!(agent = %approval.agent, error = ?e, "forge: push_config after apply failed");
}
if is_first_spawn && result.is_ok() {
// First-spawn bookkeeping: create the per-agent forge user,
// mirror the applied repo into agent-configs/<n>, and grant
// read access to core/meta.
if let Err(e) = crate::forge::ensure_user_for(&approval.agent).await {
tracing::warn!(agent = %approval.agent, error = ?e, "forge: ensure_user after first spawn failed");
}
if let Err(e) = crate::forge::ensure_config_repo(&approval.agent).await {
tracing::warn!(agent = %approval.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(&approval.agent, &core_token).await {
tracing::warn!(agent = %approval.agent, error = ?e, "forge: meta_read_access after first spawn failed");
}
if let Err(e) = crate::forge::ensure_meta_remote(&approval.agent).await {
tracing::warn!(agent = %approval.agent, error = ?e, "forge: ensure_meta_remote after first spawn failed");
}
// New container row appeared — rescan so the dashboard
// reflects the post-spawn state without a manual refetch.
coord.rescan_containers_and_emit().await;
crate::dashboard::emit_tombstones_snapshot(&coord).await;
}
finish_approval(&coord, &approval, result, terminal_tag, is_first_spawn)
approve_apply_commit(coord, approval, agent_dir, applied_dir, claude_dir, notes_dir).await
}
ApprovalKind::InitConfig => {
// Seed the proposed config repo. Runs synchronously — it's just
// a few git operations with no nixos-container involvement.
let result: Result<()> = async {
lifecycle::setup_proposed(&proposed_dir, &approval.agent).await?;
lifecycle::ensure_claude_dir(&claude_dir)?;
lifecycle::ensure_state_dir(&notes_dir)?;
Ok(())
}
.await;
// Wire the meta remote now that the proposed repo exists.
if result.is_ok()
&& let Err(e) = crate::forge::ensure_meta_remote(&approval.agent).await {
tracing::warn!(agent = %approval.agent, error = ?e, "forge: ensure_meta_remote after init_config failed");
}
finish_approval(&coord, &approval, result, None, false)
}
ApprovalKind::UpdateMetaInputs => {
// Decode the inputs from the commit_ref field (stored as JSON
// by submit_apply_commit's counterpart in manager_server.rs).
let inputs: Vec<String> =
serde_json::from_str(&approval.commit_ref).unwrap_or_default();
let result = crate::meta::lock_update(&inputs).await;
finish_approval(&coord, &approval, result, None, false)
approve_init_config(coord, approval, proposed_dir, claude_dir, notes_dir).await
}
ApprovalKind::UpdateMetaInputs => approve_update_meta_inputs(coord, approval).await,
ApprovalKind::Spawn => {
// Run the spawn in the background so the approve POST returns
// immediately. The dashboard reads `transient` to render a spinner.
// Guard is created synchronously here (so the spinner appears
// the moment the operator clicks approve) and moved into the
// task; it auto-clears even if the runtime drops the task.
let coord_bg = coord.clone();
let approval_bg = approval.clone();
let guard = coord_bg.transient_guard(&approval_bg.agent, TransientKind::Spawning);
tokio::spawn(async move {
let guard = guard;
let agent_bg = approval_bg.agent.clone();
let result = lifecycle::spawn(
&approval_bg.agent,
&coord_bg.hyperhive_flake,
&agent_dir,
&proposed_dir,
&applied_dir,
&claude_dir,
&notes_dir,
coord_bg.dashboard_port,
&coord_bg.operator_pronouns,
&coord_bg.context_window_tokens,
)
.await;
drop(guard);
if result.is_ok() {
if let Err(e) = crate::forge::ensure_user_for(&agent_bg).await {
tracing::warn!(agent = %agent_bg, error = ?e, "forge: ensure_user after spawn failed");
}
// Create the agent-configs mirror repo and seed it
// with the freshly-initialised applied repo (main +
// deployed/0).
if let Err(e) = crate::forge::ensure_config_repo(&agent_bg).await {
tracing::warn!(agent = %agent_bg, error = ?e, "forge: ensure_config_repo after spawn failed");
}
if let Err(e) = crate::forge::push_config(&agent_bg).await {
tracing::warn!(agent = %agent_bg, error = ?e, "forge: push_config after spawn failed");
}
if let Some(core_token) = crate::forge::core_token()
&& let Err(e) = crate::forge::meta_read_access(&agent_bg, &core_token).await {
tracing::warn!(agent = %agent_bg, error = ?e, "forge: meta_read_access after spawn failed");
}
if let Err(e) = crate::forge::ensure_meta_remote(&agent_bg).await {
tracing::warn!(agent = %agent_bg, error = ?e, "forge: ensure_meta_remote after spawn failed");
}
}
if let Err(e) = finish_approval(&coord_bg, &approval_bg, result, None, false) {
tracing::warn!(agent = %agent_bg, error = ?e, "spawn approval failed");
}
// New container row appeared (or didn't, on failure
// before nixos-container create completed) — rescan so
// dashboards reflect the post-spawn state. Spawn can
// also consume a tombstone of the same name; emit the
// fresh list so the operator's dormant-state pane
// updates without a refetch.
coord_bg.rescan_containers_and_emit().await;
crate::dashboard::emit_tombstones_snapshot(&coord_bg).await;
});
approve_spawn(&coord, &approval, agent_dir, proposed_dir, applied_dir, claude_dir, notes_dir);
Ok(())
}
}
}
async fn approve_apply_commit(
coord: Arc<Coordinator>,
approval: hive_sh4re::Approval,
agent_dir: std::path::PathBuf,
applied_dir: std::path::PathBuf,
claude_dir: std::path::PathBuf,
notes_dir: std::path::PathBuf,
) -> Result<()> {
let (result, terminal_tag, is_first_spawn) = run_apply_commit(
&coord,
&approval,
&agent_dir,
&applied_dir,
&claude_dir,
&notes_dir,
)
.await;
if let Err(e) = crate::forge::push_config(&approval.agent).await {
tracing::warn!(agent = %approval.agent, error = ?e, "forge: push_config after apply failed");
}
if is_first_spawn && result.is_ok() {
forge_after_first_spawn(&coord, &approval.agent).await;
}
finish_approval(&coord, &approval, result, terminal_tag, is_first_spawn)
}
/// Forge bookkeeping run once after the very first container spawn:
/// create the per-agent forge user, mirror the applied repo, and grant
/// read access to core/meta. Also rescans containers so the dashboard
/// reflects the post-spawn state.
async fn forge_after_first_spawn(coord: &Arc<Coordinator>, agent: &str) {
if let Err(e) = crate::forge::ensure_user_for(agent).await {
tracing::warn!(%agent, error = ?e, "forge: ensure_user after first spawn failed");
}
if let Err(e) = crate::forge::ensure_config_repo(agent).await {
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");
}
if let Err(e) = crate::forge::ensure_meta_remote(agent).await {
tracing::warn!(%agent, error = ?e, "forge: ensure_meta_remote after first spawn failed");
}
coord.rescan_containers_and_emit().await;
crate::dashboard::emit_tombstones_snapshot(coord).await;
}
async fn approve_init_config(
coord: Arc<Coordinator>,
approval: hive_sh4re::Approval,
proposed_dir: std::path::PathBuf,
claude_dir: std::path::PathBuf,
notes_dir: std::path::PathBuf,
) -> Result<()> {
// Seed the proposed config repo — just git operations, no nixos-container.
let result: Result<()> = async {
lifecycle::setup_proposed(&proposed_dir, &approval.agent).await?;
lifecycle::ensure_claude_dir(&claude_dir)?;
lifecycle::ensure_state_dir(&notes_dir)?;
Ok(())
}
.await;
if result.is_ok()
&& let Err(e) = crate::forge::ensure_meta_remote(&approval.agent).await {
tracing::warn!(agent = %approval.agent, error = ?e, "forge: ensure_meta_remote after init_config failed");
}
finish_approval(&coord, &approval, result, None, false)
}
async fn approve_update_meta_inputs(
coord: Arc<Coordinator>,
approval: hive_sh4re::Approval,
) -> Result<()> {
// Inputs stored as JSON in commit_ref by the manager's submit path.
let inputs: Vec<String> = serde_json::from_str(&approval.commit_ref).unwrap_or_default();
let result = crate::meta::lock_update(&inputs).await;
finish_approval(&coord, &approval, result, None, false)
}
fn approve_spawn(
coord: &Arc<Coordinator>,
approval: &hive_sh4re::Approval,
agent_dir: std::path::PathBuf,
proposed_dir: std::path::PathBuf,
applied_dir: std::path::PathBuf,
claude_dir: std::path::PathBuf,
notes_dir: std::path::PathBuf,
) {
// Run spawn in the background so approve POST returns immediately.
// Guard created synchronously so the spinner appears the moment
// the operator clicks approve; auto-clears when the task drops it.
let coord_bg = Arc::clone(coord);
let approval_bg = approval.clone();
let guard = coord_bg.transient_guard(&approval_bg.agent, TransientKind::Spawning);
tokio::spawn(async move {
let guard = guard;
let agent_bg = approval_bg.agent.clone();
let result = lifecycle::spawn(
&approval_bg.agent,
&coord_bg.hyperhive_flake,
&agent_dir,
&proposed_dir,
&applied_dir,
&claude_dir,
&notes_dir,
coord_bg.dashboard_port,
&coord_bg.operator_pronouns,
&coord_bg.context_window_tokens,
)
.await;
drop(guard);
if result.is_ok() {
if let Err(e) = crate::forge::ensure_user_for(&agent_bg).await {
tracing::warn!(agent = %agent_bg, error = ?e, "forge: ensure_user after spawn failed");
}
if let Err(e) = crate::forge::ensure_config_repo(&agent_bg).await {
tracing::warn!(agent = %agent_bg, error = ?e, "forge: ensure_config_repo after spawn failed");
}
if let Err(e) = crate::forge::push_config(&agent_bg).await {
tracing::warn!(agent = %agent_bg, error = ?e, "forge: push_config after spawn failed");
}
if let Some(core_token) = crate::forge::core_token()
&& let Err(e) = crate::forge::meta_read_access(&agent_bg, &core_token).await {
tracing::warn!(agent = %agent_bg, error = ?e, "forge: meta_read_access after spawn failed");
}
if let Err(e) = crate::forge::ensure_meta_remote(&agent_bg).await {
tracing::warn!(agent = %agent_bg, error = ?e, "forge: ensure_meta_remote after spawn failed");
}
}
if let Err(e) = finish_approval(&coord_bg, &approval_bg, result, None, false) {
tracing::warn!(agent = %agent_bg, error = ?e, "spawn approval failed");
}
coord_bg.rescan_containers_and_emit().await;
crate::dashboard::emit_tombstones_snapshot(&coord_bg).await;
});
}
fn finish_approval(
coord: &Coordinator,
approval: &hive_sh4re::Approval,
@ -268,6 +290,7 @@ fn finish_approval(
/// and reset the working tree back to the last known-good main. main
/// never advances on a failed build, so a crash-and-recover doesn't
/// leave the agent pointing at a tree it can't evaluate.
#[allow(clippy::too_many_lines)] // sequential build/tag/notify pipeline; splitting would obscure the flow
async fn run_apply_commit(
coord: &Arc<Coordinator>,
approval: &hive_sh4re::Approval,

5
hive-c0re/src/broker.rs
View file

@ -613,8 +613,9 @@ impl Broker {
let now = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.ok()
.map_or(0, |d| d.as_secs() as i64);
now - since_secs as i64
.and_then(|d| i64::try_from(d.as_secs()).ok())
.unwrap_or(0);
now.saturating_sub(i64::try_from(since_secs).unwrap_or(i64::MAX))
} else {
i64::MIN
};

4
hive-c0re/src/container_view.rs
View file

@ -164,8 +164,8 @@ fn is_rate_limited(name: &str) -> bool {
/// silently yields `None` so a missing/corrupt file never blocks
/// `build_all`.
///
/// Context tokens = `last_input_tokens + last_cache_read_input_tokens
/// + last_cache_creation_input_tokens`, mirroring
/// Context tokens are the sum of `last_input_tokens`, `last_cache_read_input_tokens`,
/// and `last_cache_creation_input_tokens`, mirroring
/// `hive_ag3nt::events::TokenUsage::context_tokens`.
fn read_last_ctx_tokens(name: &str) -> Option<u64> {
let path = Coordinator::agent_notes_dir(name).join("hyperhive-turn-stats.sqlite");

2
hive-c0re/src/events_vacuum.rs
View file

@ -23,7 +23,7 @@ const KEEP_SECS: i64 = 7 * 24 * 3600;
/// Background loop: sweep every existing agent state dir hourly, run
/// the vacuum SQL against its events.sqlite if present. Errors are
/// logged but don't tear the loop down.
pub fn spawn(coord: Arc<Coordinator>) {
pub fn spawn(coord: &Arc<Coordinator>) {
let mut shutdown = coord.shutdown_rx();
tokio::spawn(async move {
loop {

126
hive-c0re/src/lifecycle.rs
View file

@ -767,69 +767,6 @@ async fn systemd_daemon_reload() -> Result<()> {
Ok(())
}
#[cfg(test)]
mod tests {
use super::*;
/// Regression test: setup_proposed must seed both agent.nix and flake.nix
/// in the initial commit. Before commit 5b5a93e flake.nix was missing from
/// the scaffold, requiring manual creation (seen with the damocles agent).
#[tokio::test]
async fn setup_proposed_seeds_flake_nix() {
let dir = tempfile::tempdir().expect("tempdir");
let proposed = dir.path().join("proposed");
setup_proposed(&proposed, "test-agent")
.await
.expect("setup_proposed");
// Both files must exist on disk.
assert!(proposed.join("agent.nix").exists(), "agent.nix missing");
assert!(proposed.join("flake.nix").exists(), "flake.nix missing");
// flake.nix must export nixosModules.default (the meta-flake contract).
let flake = std::fs::read_to_string(proposed.join("flake.nix")).unwrap();
assert!(
flake.contains("nixosModules.default"),
"flake.nix does not export nixosModules.default"
);
// Both files must be tracked in the initial git commit.
let out = git_command()
.current_dir(&proposed)
.args(["show", "--name-only", "--format=", "HEAD"])
.output()
.await
.expect("git show");
let tracked = String::from_utf8_lossy(&out.stdout);
assert!(tracked.contains("agent.nix"), "agent.nix not committed");
assert!(tracked.contains("flake.nix"), "flake.nix not committed");
}
/// setup_proposed is idempotent: calling it on an existing repo is a
/// no-op (the fresh guard skips all writes).
#[tokio::test]
async fn setup_proposed_idempotent() {
let dir = tempfile::tempdir().expect("tempdir");
let proposed = dir.path().join("proposed");
setup_proposed(&proposed, "test-agent")
.await
.expect("first call");
// Second call must not error even though .git already exists.
setup_proposed(&proposed, "test-agent")
.await
.expect("second call");
// Still one commit.
let out = git_command()
.current_dir(&proposed)
.args(["rev-list", "--count", "HEAD"])
.output()
.await
.expect("git rev-list");
let count = String::from_utf8_lossy(&out.stdout).trim().to_owned();
assert_eq!(count, "1", "expected exactly one commit after idempotent call");
}
}
/// Idempotently rewrite the lines in `/etc/nixos-containers/<container>.conf`
/// that hive-c0re owns: `PRIVATE_NETWORK` (forced 0 so the agent's web UI port
/// is reachable on the host) and `EXTRA_NSPAWN_FLAGS` (the runtime-dir bind).
@ -1097,3 +1034,66 @@ async fn container_journal_tail(args: &[&str]) -> String {
_ => String::new(),
}
}
#[cfg(test)]
mod tests {
use super::*;
/// Regression test: `setup_proposed` must seed both agent.nix and flake.nix
/// in the initial commit. Before commit 5b5a93e flake.nix was missing from
/// the scaffold, requiring manual creation (seen with the damocles agent).
#[tokio::test]
async fn setup_proposed_seeds_flake_nix() {
let dir = tempfile::tempdir().expect("tempdir");
let proposed = dir.path().join("proposed");
setup_proposed(&proposed, "test-agent")
.await
.expect("setup_proposed");
// Both files must exist on disk.
assert!(proposed.join("agent.nix").exists(), "agent.nix missing");
assert!(proposed.join("flake.nix").exists(), "flake.nix missing");
// flake.nix must export nixosModules.default (the meta-flake contract).
let flake = std::fs::read_to_string(proposed.join("flake.nix")).unwrap();
assert!(
flake.contains("nixosModules.default"),
"flake.nix does not export nixosModules.default"
);
// Both files must be tracked in the initial git commit.
let out = git_command()
.current_dir(&proposed)
.args(["show", "--name-only", "--format=", "HEAD"])
.output()
.await
.expect("git show");
let tracked = String::from_utf8_lossy(&out.stdout);
assert!(tracked.contains("agent.nix"), "agent.nix not committed");
assert!(tracked.contains("flake.nix"), "flake.nix not committed");
}
/// `setup_proposed` is idempotent: calling it on an existing repo is a
/// no-op (the fresh guard skips all writes).
#[tokio::test]
async fn setup_proposed_idempotent() {
let dir = tempfile::tempdir().expect("tempdir");
let proposed = dir.path().join("proposed");
setup_proposed(&proposed, "test-agent")
.await
.expect("first call");
// Second call must not error even though .git already exists.
setup_proposed(&proposed, "test-agent")
.await
.expect("second call");
// Still one commit.
let out = git_command()
.current_dir(&proposed)
.args(["rev-list", "--count", "HEAD"])
.output()
.await
.expect("git rev-list");
let count = String::from_utf8_lossy(&out.stdout).trim().to_owned();
assert_eq!(count, "1", "expected exactly one commit after idempotent call");
}
}

242
hive-c0re/src/main.rs
View file

@ -116,121 +116,7 @@ async fn main() -> Result<()> {
dashboard_port,
operator_pronouns,
context_window_tokens,
} => {
let cwt: std::collections::HashMap<String, u64> =
serde_json::from_str(&context_window_tokens)
.context("--context-window-tokens: invalid JSON")?;
let coord = Arc::new(Coordinator::open(
&db,
hyperhive_flake,
dashboard_port,
operator_pronouns,
cwt,
)?);
manager_server::start(coord.clone())?;
// Idempotent pre-flight: rewrite pre-meta-layout applied
// repos, ensure proposed repos carry the `applied`
// remote, bootstrap the meta repo, repoint containers at
// `meta#<name>` (one-shot, guarded by a marker file).
// Runs before manager auto-spawn so the new manager is
// built against meta from the first attempt.
if let Err(e) = migrate::run(&coord).await {
tracing::warn!(error = ?e, "startup migration failed");
}
// Auto-create the manager container if it isn't there yet. Block
// on this — without hm1nd the system has no manager harness.
// Failures are logged but allowed: a broken auto-spawn shouldn't
// make the dashboard unreachable for debugging.
if let Err(e) = auto_update::ensure_manager(&coord).await {
tracing::warn!(error = ?e, "auto-spawn manager failed");
}
// Auto-update in the background — don't block service start.
// Sub-agent rebuilds can take tens of seconds; we want the admin
// socket up immediately.
let update_coord = coord.clone();
tokio::spawn(async move {
if let Err(e) = auto_update::run(update_coord).await {
tracing::warn!(error = ?e, "auto-update task failed");
}
});
// Forge user sweep: ensure every existing container has a
// forgejo user + access token. No-op when the hive-forge
// container isn't running. Backgrounded — touches the
// forge state dir via `nixos-container run` which is slow.
tokio::spawn(async move {
forge::ensure_all().await;
});
// Periodic broker vacuum: drop fully-acked messages older
// than 30 days. Delivered-but-unacked rows (recoverable via
// requeue_inflight) and undelivered rows are always kept.
// Runs hourly; first sweep happens immediately.
let vacuum_coord = coord.clone();
let mut vacuum_shutdown = coord.shutdown_rx();
tokio::spawn(async move {
let interval = std::time::Duration::from_secs(3600);
let keep_secs: i64 = 30 * 24 * 3600;
loop {
match vacuum_coord.broker.vacuum_delivered(keep_secs) {
Ok(0) => {}
Ok(n) => tracing::info!(removed = n, "broker vacuum"),
Err(e) => tracing::warn!(error = ?e, "broker vacuum failed"),
}
tokio::select! {
() = tokio::time::sleep(interval) => {}
_ = vacuum_shutdown.changed() => {
tracing::info!("broker vacuum: shutdown signal received");
break;
}
}
}
});
// Per-agent events.sqlite vacuum: host-side so the harness
// doesn't need any retention wiring of its own.
events_vacuum::spawn(coord.clone());
// Per-agent turn-stats.sqlite vacuum: same pattern, 90-day
// retention so trend analysis has enough history.
stats_vacuum::spawn(coord.clone());
// Container crash watcher: emits HelperEvent::ContainerCrash
// when a previously-running container goes away without an
// operator-initiated transient state.
crash_watch::spawn(coord.clone());
// Reminder scheduler: drains due reminders + handles
// file_path payload persistence. See reminder_scheduler.rs.
reminder_scheduler::spawn(coord.clone());
// Forward every broker event onto the unified dashboard
// channel with a freshly-stamped seq, so the dashboard SSE
// sees broker messages + future mutation events on one
// stream with one monotonic seq. The broker's intra-process
// channel (used by `recv_blocking_batch`) stays untouched.
spawn_broker_to_dashboard_forwarder(coord.clone());
let dash_coord = coord.clone();
tokio::spawn(async move {
if let Err(e) = dashboard::serve(dashboard_port, dash_coord).await {
tracing::error!(error = ?e, "dashboard failed");
}
});
// Run the admin socket until a signal arrives; then signal
// all background tasks so they exit cleanly before the
// process terminates.
let coord_sig = coord.clone();
tokio::select! {
res = server::serve(&cli.socket, coord) => { res? }
_ = tokio::signal::ctrl_c() => {
tracing::info!("SIGINT received — requesting shutdown");
coord_sig.request_shutdown();
}
() = async {
let mut sig = tokio::signal::unix::signal(
tokio::signal::unix::SignalKind::terminate()
).expect("failed to install SIGTERM handler");
sig.recv().await;
} => {
tracing::info!("SIGTERM received — requesting shutdown");
coord_sig.request_shutdown();
}
}
Ok(())
}
} => cmd_serve(hyperhive_flake, db, dashboard_port, operator_pronouns, context_window_tokens, &cli.socket).await,
Cmd::Spawn { name } => {
render(client::request(&cli.socket, HostRequest::Spawn { name }).await?)
}
@ -255,6 +141,132 @@ async fn main() -> Result<()> {
}
}
/// Start the coordinator daemon: open the broker, run migrations, spawn
/// background tasks (auto-update, vacuums, crash-watcher, reminder-scheduler,
/// dashboard), then serve the admin socket until a signal arrives.
async fn cmd_serve(
hyperhive_flake: String,
db: std::path::PathBuf,
dashboard_port: u16,
operator_pronouns: String,
context_window_tokens: String,
socket: &std::path::Path,
) -> Result<()> {
let cwt: std::collections::HashMap<String, u64> =
serde_json::from_str(&context_window_tokens)
.context("--context-window-tokens: invalid JSON")?;
let coord = Arc::new(Coordinator::open(
&db,
hyperhive_flake,
dashboard_port,
operator_pronouns,
cwt,
)?);
manager_server::start(coord.clone())?;
// Idempotent pre-flight: rewrite pre-meta-layout applied
// repos, ensure proposed repos carry the `applied`
// remote, bootstrap the meta repo, repoint containers at
// `meta#<name>` (one-shot, guarded by a marker file).
// Runs before manager auto-spawn so the new manager is
// built against meta from the first attempt.
if let Err(e) = migrate::run(&coord).await {
tracing::warn!(error = ?e, "startup migration failed");
}
// Auto-create the manager container if it isn't there yet. Block
// on this — without hm1nd the system has no manager harness.
// Failures are logged but allowed: a broken auto-spawn shouldn't
// make the dashboard unreachable for debugging.
if let Err(e) = auto_update::ensure_manager(&coord).await {
tracing::warn!(error = ?e, "auto-spawn manager failed");
}
// Auto-update in the background — don't block service start.
// Sub-agent rebuilds can take tens of seconds; we want the admin
// socket up immediately.
let update_coord = coord.clone();
tokio::spawn(async move {
if let Err(e) = auto_update::run(update_coord).await {
tracing::warn!(error = ?e, "auto-update task failed");
}
});
// Forge user sweep: ensure every existing container has a
// forgejo user + access token. No-op when the hive-forge
// container isn't running. Backgrounded — touches the
// forge state dir via `nixos-container run` which is slow.
tokio::spawn(async move {
forge::ensure_all().await;
});
// Periodic broker vacuum: drop fully-acked messages older
// than 30 days. Delivered-but-unacked rows (recoverable via
// requeue_inflight) and undelivered rows are always kept.
// Runs hourly; first sweep happens immediately.
let vacuum_coord = coord.clone();
let mut vacuum_shutdown = coord.shutdown_rx();
tokio::spawn(async move {
let interval = std::time::Duration::from_secs(3600);
let keep_secs: i64 = 30 * 24 * 3600;
loop {
match vacuum_coord.broker.vacuum_delivered(keep_secs) {
Ok(0) => {}
Ok(n) => tracing::info!(removed = n, "broker vacuum"),
Err(e) => tracing::warn!(error = ?e, "broker vacuum failed"),
}
tokio::select! {
() = tokio::time::sleep(interval) => {}
_ = vacuum_shutdown.changed() => {
tracing::info!("broker vacuum: shutdown signal received");
break;
}
}
}
});
// Per-agent events.sqlite vacuum: host-side so the harness
// doesn't need any retention wiring of its own.
events_vacuum::spawn(&coord);
// Per-agent turn-stats.sqlite vacuum: same pattern, 90-day
// retention so trend analysis has enough history.
stats_vacuum::spawn(&coord);
// Container crash watcher: emits HelperEvent::ContainerCrash
// when a previously-running container goes away without an
// operator-initiated transient state.
crash_watch::spawn(coord.clone());
// Reminder scheduler: drains due reminders + handles
// file_path payload persistence. See reminder_scheduler.rs.
reminder_scheduler::spawn(coord.clone());
// Forward every broker event onto the unified dashboard
// channel with a freshly-stamped seq, so the dashboard SSE
// sees broker messages + future mutation events on one
// stream with one monotonic seq. The broker's intra-process
// channel (used by `recv_blocking_batch`) stays untouched.
spawn_broker_to_dashboard_forwarder(coord.clone());
let dash_coord = coord.clone();
tokio::spawn(async move {
if let Err(e) = dashboard::serve(dashboard_port, dash_coord).await {
tracing::error!(error = ?e, "dashboard failed");
}
});
// Run the admin socket until a signal arrives; then signal
// all background tasks so they exit cleanly before the
// process terminates.
let coord_sig = coord.clone();
tokio::select! {
res = server::serve(socket, coord) => { res? }
_ = tokio::signal::ctrl_c() => {
tracing::info!("SIGINT received — requesting shutdown");
coord_sig.request_shutdown();
}
() = async {
let mut sig = tokio::signal::unix::signal(
tokio::signal::unix::SignalKind::terminate()
).expect("failed to install SIGTERM handler");
sig.recv().await;
} => {
tracing::info!("SIGTERM received — requesting shutdown");
coord_sig.request_shutdown();
}
}
Ok(())
}
/// Re-emit every broker `MessageEvent` onto the dashboard channel as
/// a `DashboardEvent::Sent` / `Delivered` with a freshly-stamped seq.
/// Background task; runs for the life of the process. On a lagged

2
hive-c0re/src/stats_vacuum.rs
View file

@ -21,7 +21,7 @@ const KEEP_SECS: i64 = 90 * 24 * 3600;
/// Background loop: sweep every existing agent state dir hourly, run
/// the vacuum SQL against its turn-stats.sqlite if present. Errors
/// are logged but don't tear the loop down.
pub fn spawn(coord: Arc<Coordinator>) {
pub fn spawn(coord: &Arc<Coordinator>) {
let mut shutdown = coord.shutdown_rx();
tokio::spawn(async move {
loop {

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

@ -133,6 +133,7 @@ pub struct ReminderStats {
}
impl HostResponse {
#[must_use]
pub fn success() -> Self {
Self {
ok: true,
@ -142,6 +143,7 @@ impl HostResponse {
}
}
#[must_use]
pub fn error(message: impl Into<String>) -> Self {
Self {
ok: false,
@ -151,6 +153,7 @@ impl HostResponse {
}
}
#[must_use]
pub fn list(agents: Vec<String>) -> Self {
Self {
ok: true,
@ -160,6 +163,7 @@ impl HostResponse {
}
}
#[must_use]
pub fn pending(approvals: Vec<Approval>) -> Self {
Self {
ok: true,