From 001ea38ea40e1468ddd7386d4e1c9d243c2528ec Mon Sep 17 00:00:00 2001 From: damocles Date: Sat, 25 Jul 2026 20:44:27 +0200 Subject: [PATCH] gate stale todo wakes on an emptiness check (#2678) --- hive-agent/src/main.rs | 34 +++++++++++++++++++++++++++++++--- hive-agent/src/todos.rs | 34 ++++++++++++++++++++++++++++++++++ 2 files changed, 65 insertions(+), 3 deletions(-) diff --git a/hive-agent/src/main.rs b/hive-agent/src/main.rs index 0c7f2e6d..78efa744 100644 --- a/hive-agent/src/main.rs +++ b/hive-agent/src/main.rs @@ -500,22 +500,32 @@ async fn serve_main(socket: &Path, poll_ms: u64) -> Result<()> { // acceptable since a from-scratch harness boot either has a writable // harness dir or doesn't). let todo_wake = Arc::new(tokio::sync::Notify::new()); - match todos::Todos::open(&paths::state_db()) { + // Kept alongside `todo_wake` so the serve loop's `LocalTodo` arm can + // gate a wake on `has_any()` before spawning a turn — see its doc + // comment (the phantom-todo-wake issue: a burst of same-turn upserts + // can arm a second `Notify` permit that outlives the turn that + // already drained its payload). + let todos_store: Option> = match todos::Todos::open(&paths::state_db()) { Ok(store) => { let store = Arc::new(store); let wake = todo_wake.clone(); let reminders = reminder_store.clone(); let bus_for_socket = bus.clone(); + let store_for_socket = store.clone(); tokio::spawn(async move { - if let Err(e) = todo_server::run(store, wake, reminders, bus_for_socket).await { + if let Err(e) = + todo_server::run(store_for_socket, wake, reminders, bus_for_socket).await + { tracing::error!(error = %e, "in-agent todo socket exited with error"); } }); + Some(store) } Err(e) => { tracing::error!(error = ?e, "open todos db failed — in-agent todo socket disabled"); + None } - } + }; if matches!(initial, LoginState::NeedsLogin) { login::wait_for_login(&claude_dir, login_state.clone(), &bus, poll_ms).await; } else { @@ -533,6 +543,7 @@ async fn serve_main(socket: &Path, poll_ms: u64) -> Result<()> { stats, &files, todo_wake, + todos_store, reminder_rx, ) .await @@ -556,6 +567,7 @@ async fn serve_loop( stats: Option, files: &turn::TurnFiles, todo_wake: Arc, + todos_store: Option>, mut reminder_rx: tokio::sync::mpsc::UnboundedReceiver, ) -> Result<()> { tracing::info!(socket = %socket.display(), "harness serve"); @@ -586,6 +598,22 @@ async fn serve_loop( } { RecvOutcome::Message(first) => first, RecvOutcome::LocalTodo => { + // Gate on `has_any()` before spawning a turn: a burst of + // same-turn upserts can arm a second `Notify` permit that + // outlives the turn which already drained its payload + // (the phantom-todo-wake issue) — `notify_one` doesn't + // coalesce once the first permit's been consumed, so the surplus wake + // fires the instant the loop is back here even though + // there's nothing left to show. Fail open (drive a turn + // anyway) on a `has_any` error so a flaky sqlite read + // never silently swallows a real wake. + let has_any = todos_store + .as_ref() + .is_none_or(|store| store.has_any().unwrap_or(true)); + if !has_any { + tracing::debug!("todo wake fired against an empty store — stale, skipping"); + continue; + } tracing::debug!("todo wake consumed, sending synthetic todo message"); synthetic_todo_message() } diff --git a/hive-agent/src/todos.rs b/hive-agent/src/todos.rs index 19af3142..a5d7da2f 100644 --- a/hive-agent/src/todos.rs +++ b/hive-agent/src/todos.rs @@ -223,6 +223,27 @@ impl Todos { .collect::>>()?; Ok(rows) } + + /// Cheap existence check — `true` when at least one todo row exists + /// (across every subsystem). Used to gate a `todo_wake` notification + /// against being turned into a turn when its payload has already been + /// drained by an earlier turn (see the serve loop's `LocalTodo` arm): + /// an `EXISTS` probe, not a full `list` + row materialization, since + /// this runs on every wake. + /// + /// # Errors + /// + /// Propagates the sqlite query failure. + /// + /// # Panics + /// + /// Panics if the connection mutex is poisoned. + pub fn has_any(&self) -> Result { + let conn = self.conn.lock().unwrap(); + let any: bool = + conn.query_row("SELECT EXISTS(SELECT 1 FROM todos)", [], |row| row.get(0))?; + Ok(any) + } } #[cfg(test)] @@ -279,6 +300,19 @@ mod tests { assert!(s.list(None).unwrap().is_empty()); } + #[test] + fn has_any_reflects_emptiness() { + let (_dir, s) = store(); + assert!(!s.has_any().unwrap(), "fresh store has no todos"); + let (id, _) = s.upsert("bash", None, "task done", None).unwrap(); + assert!(s.has_any().unwrap()); + s.mark_done(id).unwrap(); + assert!( + !s.has_any().unwrap(), + "empty again after draining the only row" + ); + } + #[test] fn clear_subsystem_wipes_only_its_own() { let (_dir, s) = store();