auto-pause the turn loop after repeated missed get_loose_ends on todo wakes

hive-agent/src/main.rs

@@ -51,6 +51,16 @@ const DEFAULT_WEB_PORT: u16 = 8042;
 /// `hivectl resume` feeling immediate.
 const PAUSE_POLL: Duration = Duration::from_secs(5);
 
+/// Consecutive todo-wake turns that skip `get_loose_ends` before the serve
+/// loop pauses itself (writes the same `paused_marker()` file `hivectl
+/// agent <name> pause`/`resume` already toggle — no new plumbing). A
+/// deliberately small number: a genuine miss should be rare, and pausing
+/// quickly beats letting an unacked todo backlog balloon to a size that
+/// makes `get_loose_ends` itself expensive/unwieldy to read (the failure
+/// mode that motivated this in the first place — a reviewer's call on the
+/// exact threshold, adjustable if it proves too twitchy in practice).
+const TODO_MISS_PAUSE_THRESHOLD: u32 = 3;
+
 use std::path::{Path, PathBuf};
 use std::sync::{Arc, Mutex};
 use std::time::Duration;
@@ -192,6 +202,12 @@ fn format_turn_failure(err: &anyhow::Error) -> String {
 struct TurnControl {
     /// The turn ended in `AuthFailed` — caller parks on login.
     auth_failed: bool,
+    /// `Some(called)` when this turn was driven by a todo wake (`from ==
+    /// "todo"`) — `called` is whether the turn actually invoked
+    /// `mcp__hyperhive__get_loose_ends` at some point. `None` for every
+    /// other wake source, so the serve loop's miss-streak only reacts to
+    /// todo-driven turns. See `TODO_MISS_PAUSE_THRESHOLD`.
+    todo_wake_checked: Option<bool>,
 }
 
 /// Synthesize the message that drives a turn when an in-container producer
@@ -201,10 +217,23 @@ struct TurnControl {
 /// marker file). `id = 0` is a non-broker sentinel: the synthetic message
 /// has no DB row, and `AckTurn` keys off the recipient's in-flight list
 /// (which is empty here) rather than this id.
-fn synthetic_todo_message() -> hive_sh4re::DeliveredMessage {
+///
+/// `stern` renders a harsher final-warning body instead of the usual one —
+/// set when this is the last todo wake before the miss-streak would hit
+/// `TODO_MISS_PAUSE_THRESHOLD`, so the agent gets one unambiguous chance to
+/// avoid being auto-paused.
+fn synthetic_todo_message(stern: bool) -> hive_sh4re::DeliveredMessage {
+    let body = if stern {
+        "you have todos — call get_loose_ends NOW. you've skipped it on recent todo \
+         wakes in a row; if this turn doesn't call it, the harness will pause your \
+         own turn loop until an operator resumes you."
+            .to_owned()
+    } else {
+        "you have todos — call get_loose_ends to see them".to_owned()
+    };
     hive_sh4re::DeliveredMessage {
         from: "todo".into(),
-        body: "you have todos — call get_loose_ends to see them".into(),
+        body,
         id: 0,
         redelivered: false,
         in_reply_to: None,
@@ -636,6 +665,12 @@ async fn serve_loop<S: Surface>(
     // Tracks the last observed pause state so the transitions get logged
     // once each instead of twelve lines a minute while parked.
     let mut was_paused = false;
+    // Consecutive todo-wake turns in a row that skipped `get_loose_ends`.
+    // Incremented/reset by the `todo_wake_checked` signal off each turn's
+    // `TurnControl`; see `TODO_MISS_PAUSE_THRESHOLD`. Reset to 0 on resume
+    // too, so a just-unparked agent gets a clean slate rather than being
+    // one miss away from an instant re-pause.
+    let mut todo_miss_streak: u32 = 0;
     loop {
         // Pause gate. While the marker is present this loop drives no
         // turns at all.
@@ -668,6 +703,7 @@ async fn serve_loop<S: Surface>(
                 text: "resumed: draining whatever queued while paused".into(),
             });
             was_paused = false;
+            todo_miss_streak = 0;
         }
         let next = match {
             // Idle wait: race the broker long-poll against a local
@@ -701,7 +737,10 @@ async fn serve_loop<S: Surface>(
                     continue;
                 }
                 tracing::debug!("todo wake consumed, sending synthetic todo message");
-                synthetic_todo_message()
+                // If this wake goes missed too, the streak hits the pause
+                // threshold — say so up front instead of pausing silently.
+                let stern = todo_miss_streak + 1 >= TODO_MISS_PAUSE_THRESHOLD;
+                synthetic_todo_message(stern)
             }
             RecvOutcome::Empty => {
                 // Idle: no message this poll. Service a queued operator
@@ -750,6 +789,7 @@ async fn serve_loop<S: Surface>(
             &interrupted,
         )
         .await;
+        apply_todo_wake_checked(ctrl.todo_wake_checked, &mut todo_miss_streak, &bus);
         if ctrl.auth_failed {
             *login_state.lock().unwrap() = LoginState::NeedsLogin;
             login::wait_for_login(
@@ -763,10 +803,46 @@ async fn serve_loop<S: Surface>(
     }
 }
 
+/// Apply a finished turn's `todo_wake_checked` signal to the miss-streak:
+/// resets on a checked wake, increments (and auto-pauses at the threshold)
+/// on a missed one, no-ops for a non-todo-wake turn. Split out of
+/// `serve_loop` purely to keep that function under clippy's line limit —
+/// see `TODO_MISS_PAUSE_THRESHOLD`'s doc comment for the design rationale.
+fn apply_todo_wake_checked(checked: Option<bool>, todo_miss_streak: &mut u32, bus: &Bus) {
+    match checked {
+        Some(true) => *todo_miss_streak = 0,
+        Some(false) => {
+            *todo_miss_streak += 1;
+            let streak = *todo_miss_streak;
+            tracing::warn!(
+                streak,
+                threshold = TODO_MISS_PAUSE_THRESHOLD,
+                "todo wake turn ended without calling get_loose_ends"
+            );
+            if streak >= TODO_MISS_PAUSE_THRESHOLD {
+                tracing::warn!("todo-miss streak hit the threshold — pausing the turn loop");
+                bus.emit(LiveEvent::Note {
+                    text: format!(
+                        "auto-paused: skipped get_loose_ends on {streak} \
+                         consecutive todo wakes — an operator needs to resume this agent"
+                    ),
+                });
+                if let Err(e) = std::fs::write(hive_sh4re::paths::paused_marker(), "") {
+                    tracing::warn!(error = ?e, "failed to write pause marker");
+                }
+                // Fresh slate for whenever this agent gets resumed — see
+                // `serve_loop`'s `was_paused` reset for the mirror side.
+                *todo_miss_streak = 0;
+            }
+        }
+        None => {}
+    }
+}
+
 /// Drive a single turn: emit boot-of-turn events, run claude, ack on
 /// success / requeue on rate-limit-or-401 / notify parent on failure,
-/// record stats. Returns a `TurnControl` carrying the auth-failed flag —
-/// the serve loop decides what to do next.
+/// record stats. Returns a `TurnControl` carrying the auth-failed flag and
+/// the todo-wake-checked signal — the serve loop decides what to do next.
 async fn handle_turn<S: Surface>(
     socket: &Path,
     bus: &Bus,
@@ -811,6 +887,17 @@ async fn handle_turn<S: Surface>(
         S::ack_turn(socket).await;
     }
     handle_turn_error_recovery::<S>(&outcome, bus, socket).await;
+    // Single read of the per-turn tool-call counter: `take_tool_calls`
+    // resets it, so this is the only chance to see which tools this turn
+    // invoked. Used both for the stats row below (if stats are configured)
+    // and for the todo-miss-streak signal returned to the serve loop —
+    // deliberately unconditional (not gated on `stats.is_some()`) so the
+    // counter is reset every turn regardless, and so a stats-less harness
+    // (tests, or a future config without a stats sink) still gets the
+    // miss-streak signal.
+    let tool_calls = bus.take_tool_calls();
+    let todo_wake_checked =
+        (from == "todo").then(|| tool_calls.contains_key("mcp__hyperhive__get_loose_ends"));
     if let Some(stats) = stats {
         // Fresh session this turn → mint a `sessions` row and set its id on
         // the bus so this turn (and subsequent ones until the next fresh
@@ -831,6 +918,7 @@ async fn handle_turn<S: Surface>(
             wake_from: from.clone(),
             outcome: &outcome,
             bus,
+            tool_calls,
             open_threads_count: open_threads,
             open_reminders_count: open_reminders,
         });
@@ -842,6 +930,7 @@ async fn handle_turn<S: Surface>(
     }
     TurnControl {
         auth_failed: matches!(outcome, Err(turn::TurnError::AuthFailed)),
+        todo_wake_checked,
     }
 }
 

hive-agent/src/serve_common.rs

@@ -44,6 +44,10 @@ pub fn format_wake_prompt(
 
 /// Field-named args for [`build_row`]. Mirrors the turn-stats row
 /// columns; `outcome` and `bus` borrow for the duration of the call.
+/// `tool_calls` is passed in (rather than pulled from `bus` internally)
+/// because the caller needs the same map to check for a `get_loose_ends`
+/// call before it's consumed — `Bus::take_tool_calls` resets the counter,
+/// so there's only one chance to read it per turn.
 pub struct TurnRowArgs<'a> {
     pub started_at: i64,
     pub ended_at: i64,
@@ -52,6 +56,7 @@ pub struct TurnRowArgs<'a> {
     pub wake_from: String,
     pub outcome: &'a TurnOutcome,
     pub bus: &'a Bus,
+    pub tool_calls: std::collections::HashMap<String, u64>,
     pub open_threads_count: Option<u64>,
     pub open_reminders_count: Option<u64>,
 }
@@ -69,6 +74,7 @@ pub fn build_row(args: TurnRowArgs<'_>) -> TurnStatRow {
         wake_from,
         outcome,
         bus,
+        tool_calls,
         open_threads_count,
         open_reminders_count,
     } = args;
@@ -80,7 +86,6 @@ pub fn build_row(args: TurnRowArgs<'_>) -> TurnStatRow {
     let model = bus.last_resolved_model().unwrap_or(model);
     let cost = bus.last_cost_usage().unwrap_or_default();
     let ctx = bus.last_ctx_usage().unwrap_or(cost);
-    let tool_calls = bus.take_tool_calls();
     let tool_call_count: u64 = tool_calls.values().copied().sum();
     let tool_call_breakdown_json = if tool_calls.is_empty() {
         None