docs: add missing #[must_use], # Errors, # Panics across public api

refactor: split long functions per review feedback; remove all #[allow] attributes
fix: use try_from for i64/u64 casts; split format_notification into helpers
2026-05-22 19:41:27 +02:00 · 2026-05-22 19:24:44 +02:00 · 2026-05-22 19:14:22 +02:00 · 2026-05-22 19:11:40 +02:00
20 changed files with 833 additions and 643 deletions
--- a/hive-ag3nt/src/bin/hive-ag3nt.rs
+++ b/hive-ag3nt/src/bin/hive-ag3nt.rs
@ -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
+            // Explicit long-poll: park until a message arrives (180s cap).
-            // `None` as "peek, don't wait", which would tight-loop on
+            // `max: None` (= 1) — one turn per wake; claude calls
-            // sleep(interval). The harness wants to park until a
+            // recv(max: N) in-turn to drain bursts.
            // 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.
            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;
+                handle_agent_turn(socket, &bus, stats.as_ref(), files, &turn_lock, label, first).await;
                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;
            }
            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
+/// Drive one turn for a received agent-inbox message.
-/// (`prompts/agent.md` → `claude --system-prompt-file`); this is just the
+async fn handle_agent_turn(
-/// wake signal claude reacts to. `unread` is the count of *other*
+    socket: &Path,
-/// messages in the inbox right after this one was popped.
+    bus: &Bus,
-/// `redelivered` flags messages that were popped in a prior harness
+    stats: Option<&TurnStats>,
-/// session, never acked, and resurfaced after a restart — a banner
+    files: &turn::TurnFiles,
-/// at the top of the wake prompt warns that any side-effects of
+    turn_lock: &TurnLock,
-/// previous handling may already have happened.
+    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
--- a/hive-ag3nt/src/bin/hive-m1nd.rs
+++ b/hive-ag3nt/src/bin/hive-m1nd.rs
@ -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;
+                handle_manager_turn(socket, &bus, stats.as_ref(), files, &turn_lock, first).await;
                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");
                }
            }
            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
--- a/hive-ag3nt/src/client.rs
+++ b/hive-ag3nt/src/client.rs
@ -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,
--- a/hive-ag3nt/src/events.rs
+++ b/hive-ag3nt/src/events.rs
@ -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()
    }
--- a/hive-ag3nt/src/forge_notify.rs
+++ b/hive-ag3nt/src/forge_notify.rs
@ -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");
-
+    let meta_suffix = build_meta_suffix(subject.as_ref(), is_pr);
    // 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}"),
        }
    };
    // 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.
+        format_comment_notification(client, token, &meta, comment_api_url, comment_html_url, own_login).await
        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)
        }
    } else {
-        // Notification triggered by creation or state change of the subject.
+        format_state_change_notification(notif, &meta, own_login)
-        //
+    }
-        // 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.
+/// Shared notification metadata extracted from the raw Forgejo JSON.
-        // `reason == "author"` combined with open state means we just opened the
+struct NotifMeta<'a> {
-        // issue/PR. We do NOT filter merged/closed state changes — those are
+    title: &'a str,
-        // triggered by someone else and we want them.
+    notif_type: &'a str,
-        let is_new = notif_state == "open" || notif_state.is_empty();
+    html_url: &'a str,
-        if is_new && reason == "author" && !own_login.is_empty() {
+    num: String,
-            debug!(%own_login, "forge_notify: skipping self-authored new item");
+    repo: String,
-            return None;
+    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);
+/// Build the `\nassignee: ...` (and optionally `\nreviewer: ...`) suffix
-        let kind = match notif_state {
+/// appended to all notification kinds.
-            "merged" => format!("{label} merged{num}{repo}"),
+fn build_meta_suffix(subject: Option<&serde_json::Value>, is_pr: bool) -> String {
-            "closed" => format!("{label} closed{num}{repo}"),
+    let assignees: Vec<&str> = subject
-            "open" | "" => format!("new {label}{num}{repo}"),
+        .and_then(|s| s["assignees"].as_array())
-            other => format!("{label}{num}{repo}: {other}"),
+        .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
+/// Format a notification triggered by a new comment or review submission.
-        // reason == "review_requested" (observed reason is null). Check
+async fn format_comment_notification(
-        // requested_reviewers instead, which is reliable. If own_login is
+    client: &reqwest::Client,
-        // in the list, this is a review request -- override the kind.
+    token: &str,
-        // `subject` and `is_pr` are already fetched unconditionally above (#256).
+    meta: &NotifMeta<'_>,
-        let is_review_request = is_new
+    comment_api_url: &str,
-            && is_pr
+    comment_html_url: &str,
-            && !own_login.is_empty()
+    own_login: &str,
-            && subject
+) -> Option<String> {
-                .as_ref()
+    let payload = fetch_json(client, comment_api_url, token).await;
                .and_then(|s| s["requested_reviewers"].as_array())
                .map(|arr| arr.iter().any(|r| r["login"].as_str() == Some(own_login)))
                .unwrap_or(false);
-        let kind = if is_review_request {
+    let actor_login = payload
-            format!("review requested{num}{repo}")
+        .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
+            write!(out, "\n\n{author}: {}", truncate(body_text, BODY_TRUNCATE)).ok();
-        };
+        }
-
+        out.push_str(meta_suffix);
-        let mut out = format!("[{kind}] {title}\nurl: {html_url}");
+        Some(out)
-        out.push_str(&meta_suffix);
+    } 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() {
+        let Some(id) = notif["id"].as_u64() else { continue };
            Some(n) => n,
            None => 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;
        };
--- a/hive-ag3nt/src/login_session.rs
+++ b/hive-ag3nt/src/login_session.rs
@ -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() {
--- a/hive-ag3nt/src/mcp.rs
+++ b/hive-ag3nt/src/mcp.rs
@ -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?;
--- a/hive-ag3nt/src/serve_common.rs
+++ b/hive-ag3nt/src/serve_common.rs
@ -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,
--- a/hive-ag3nt/src/stats.rs
+++ b/hive-ag3nt/src/stats.rs
@ -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,
--- a/hive-ag3nt/src/turn.rs
+++ b/hive-ag3nt/src/turn.rs
@ -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(),
--- a/hive-ag3nt/src/turn_stats.rs
+++ b/hive-ag3nt/src/turn_stats.rs
@ -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,
--- a/hive-ag3nt/src/web_ui.rs
+++ b/hive-ag3nt/src/web_ui.rs
@ -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)
 }
--- a/hive-c0re/src/actions.rs
+++ b/hive-c0re/src/actions.rs
@ -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(
+            approve_apply_commit(coord, approval, agent_dir, applied_dir, claude_dir, notes_dir).await
                &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)
        }
        ApprovalKind::InitConfig => {
-            // Seed the proposed config repo. Runs synchronously — it's just
+            approve_init_config(coord, approval, proposed_dir, claude_dir, notes_dir).await
            // 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)
        }
        ApprovalKind::UpdateMetaInputs => approve_update_meta_inputs(coord, approval).await,
        ApprovalKind::Spawn => {
-            // Run the spawn in the background so the approve POST returns
+            approve_spawn(&coord, &approval, agent_dir, proposed_dir, applied_dir, claude_dir, notes_dir);
            // 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;
            });
            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,
--- a/hive-c0re/src/broker.rs
+++ b/hive-c0re/src/broker.rs
@ -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);
+                .and_then(|d| i64::try_from(d.as_secs()).ok())
-            now - since_secs as i64
+                .unwrap_or(0);
            now.saturating_sub(i64::try_from(since_secs).unwrap_or(i64::MAX))
        } else {
            i64::MIN
        };
--- a/hive-c0re/src/container_view.rs
+++ b/hive-c0re/src/container_view.rs
@ -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
+/// Context tokens are the sum of `last_input_tokens`, `last_cache_read_input_tokens`,
-/// + last_cache_creation_input_tokens`, mirroring
+/// 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");
--- a/hive-c0re/src/events_vacuum.rs
+++ b/hive-c0re/src/events_vacuum.rs
@ -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 {
--- a/hive-c0re/src/lifecycle.rs
+++ b/hive-c0re/src/lifecycle.rs
@ -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");
    }
 }
--- a/hive-c0re/src/main.rs
+++ b/hive-c0re/src/main.rs
@ -116,121 +116,7 @@ async fn main() -> Result<()> {
            dashboard_port,
            operator_pronouns,
            context_window_tokens,
-        } => {
+        } => cmd_serve(hyperhive_flake, db, dashboard_port, operator_pronouns, context_window_tokens, &cli.socket).await,
            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::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
--- a/hive-c0re/src/stats_vacuum.rs
+++ b/hive-c0re/src/stats_vacuum.rs
@ -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 {
--- a/hive-sh4re/src/lib.rs
+++ b/hive-sh4re/src/lib.rs
@ -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,
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