From edb4aa98c6b7ab3cfd62158fdce2649c0c9aed05 Mon Sep 17 00:00:00 2001 From: damocles Date: Sun, 2 Aug 2026 17:40:17 +0200 Subject: [PATCH] auto-pause the turn loop after repeated missed get_loose_ends on todo wakes --- hive-agent/src/main.rs | 99 ++++++++++++++++++++++++++++++++-- hive-agent/src/serve_common.rs | 7 ++- 2 files changed, 100 insertions(+), 6 deletions(-) diff --git a/hive-agent/src/main.rs b/hive-agent/src/main.rs index 554b8dbd..8ac8460e 100644 --- a/hive-agent/src/main.rs +++ b/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 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, } /// 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( // 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( 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( 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( &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( } } +/// 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, 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( socket: &Path, bus: &Bus, @@ -811,6 +887,17 @@ async fn handle_turn( S::ack_turn(socket).await; } handle_turn_error_recovery::(&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( 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( } TurnControl { auth_failed: matches!(outcome, Err(turn::TurnError::AuthFailed)), + todo_wake_checked, } } diff --git a/hive-agent/src/serve_common.rs b/hive-agent/src/serve_common.rs index fda3c738..a0b14e26 100644 --- a/hive-agent/src/serve_common.rs +++ b/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, pub open_threads_count: Option, pub open_reminders_count: Option, } @@ -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