diff --git a/hive-ag3nt/src/bin/hive-ag3nt.rs b/hive-ag3nt/src/bin/hive-ag3nt.rs
index 4b06199b..44c0ae66 100644
--- 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
-            // `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
diff --git a/hive-ag3nt/src/bin/hive-m1nd.rs b/hive-ag3nt/src/bin/hive-m1nd.rs
index 816ce64c..21e5f3f6 100644
--- 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;
-                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
diff --git a/hive-ag3nt/src/client.rs b/hive-ag3nt/src/client.rs
index 64e12932..5f4963be 100644
--- 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,
diff --git a/hive-ag3nt/src/events.rs b/hive-ag3nt/src/events.rs
index ac409508..7c0dff6e 100644
--- 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()
     }
diff --git a/hive-ag3nt/src/forge_notify.rs b/hive-ag3nt/src/forge_notify.rs
index b043424e..21977f73 100644
--- 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");
-
-    // 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;
         };
diff --git a/hive-ag3nt/src/login_session.rs b/hive-ag3nt/src/login_session.rs
index 68aca36c..956375be 100644
--- 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() {
diff --git a/hive-ag3nt/src/mcp.rs b/hive-ag3nt/src/mcp.rs
index 1bd52d5c..3ae9b5e2 100644
--- 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?;
diff --git a/hive-ag3nt/src/serve_common.rs b/hive-ag3nt/src/serve_common.rs
index a1c9d48a..a84dc9fe 100644
--- 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,
diff --git a/hive-ag3nt/src/stats.rs b/hive-ag3nt/src/stats.rs
index f22165fc..1731a20c 100644
--- 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,
diff --git a/hive-ag3nt/src/turn.rs b/hive-ag3nt/src/turn.rs
index db5a69c8..9301479e 100644
--- 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(),
diff --git a/hive-ag3nt/src/turn_stats.rs b/hive-ag3nt/src/turn_stats.rs
index ffae2807..a17f81ad 100644
--- 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,
diff --git a/hive-ag3nt/src/web_ui.rs b/hive-ag3nt/src/web_ui.rs
index a0d2b0bc..644844c0 100644
--- 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)
 }
 
diff --git a/hive-c0re/src/actions.rs b/hive-c0re/src/actions.rs
index 977b9f2c..36fa764a 100644
--- 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(
-                &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,
diff --git a/hive-c0re/src/broker.rs b/hive-c0re/src/broker.rs
index f2e16984..aa414ea8 100644
--- 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);
-            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
         };
diff --git a/hive-c0re/src/container_view.rs b/hive-c0re/src/container_view.rs
index ffbf5b9d..0f4c51bb 100644
--- 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
-/// + 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");
diff --git a/hive-c0re/src/events_vacuum.rs b/hive-c0re/src/events_vacuum.rs
index b545f22e..8a487cce 100644
--- 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 {
diff --git a/hive-c0re/src/lifecycle.rs b/hive-c0re/src/lifecycle.rs
index 5e34a02e..2f67992f 100644
--- 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");
+    }
+}
diff --git a/hive-c0re/src/main.rs b/hive-c0re/src/main.rs
index 70f94095..07e6d174 100644
--- 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,
-        } => {
-            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
diff --git a/hive-c0re/src/stats_vacuum.rs b/hive-c0re/src/stats_vacuum.rs
index 4f921601..c08be606 100644
--- 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 {
diff --git a/hive-sh4re/src/lib.rs b/hive-sh4re/src/lib.rs
index 199576e3..c743b939 100644
--- 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,