feat(#2569): drive a turn on a local todo signal (serve-loop select)

hive-agent-sock/src/lib.rs

@@ -1,6 +1,6 @@
 //! Wire types for the *in-agent* socket, served by the hive-agent harness
 //! to the in-container producers (matrix / bash MCP daemons) and
-//! forge_notify. Currently carries the loose-ends-v2 *todo* op family;
+//! `forge_notify`. Currently carries the loose-ends-v2 *todo* op family;
 //! more in-agent request families may be added over time (the socket is
 //! deliberately named for the agent, not the todos).
 //!

hive-agent/src/main.rs

@@ -171,6 +171,22 @@ fn synthetic_continue() -> hive_sh4re::DeliveredMessage {
     }
 }
 
+/// Synthesize the message that drives a turn when an in-container producer
+/// upserted a new/changed *todo* over the in-agent socket (loose-ends v2).
+/// The harness owns the todo store locally and signals the serve loop
+/// directly — so this wake never touches the broker (no long-poll, no
+/// marker file). `id = 0` is the same non-broker sentinel as
+/// [`synthetic_continue`].
+fn synthetic_todo_message() -> hive_sh4re::DeliveredMessage {
+    hive_sh4re::DeliveredMessage {
+        from: "todo".into(),
+        body: "you have todos — call get_loose_ends to see them".into(),
+        id: 0,
+        redelivered: false,
+        in_reply_to: None,
+    }
+}
+
 /// Synthetic message that drives the single stop-checkpoint turn when c0re
 /// signals a graceful stop. The agent gets one final turn to flush durable
 /// `/state` before the container is stopped; new inbound is already fenced.
@@ -210,6 +226,10 @@ enum RecvOutcome {
     /// one stop-checkpoint turn (flush durable `/state`), reports
     /// `GracefulStopComplete`, and exits so the container can be stopped.
     GracefulStop,
+    /// An in-container producer upserted a new/changed todo over the
+    /// in-agent socket; the serve loop drives a `synthetic_todo_message`
+    /// turn. Not a broker message — the harness signalled itself directly.
+    LocalTodo,
 }
 
 /// Wire surface abstraction. `AgentSurface` is the only impl — the trait
@@ -425,6 +445,26 @@ async fn serve_main<S: Surface>(socket: &Path, poll_ms: u64) -> Result<()> {
             tracing::error!(error = %e, "web_ui::serve exited with error");
         }
     });
+    // In-agent todo socket (loose-ends v2): the harness owns the todo store
+    // locally and serves the in-container producers on `HIVE_AGENT_SOCKET`.
+    // A new/changed upsert fires `todo_wake` so the serve loop drives a turn
+    // directly — no broker round-trip, no marker files. Best-effort: if the
+    // store can't open, the socket just isn't served.
+    let todo_wake = Arc::new(tokio::sync::Notify::new());
+    match todos::Todos::open(&paths::todos_db()) {
+        Ok(store) => {
+            let store = Arc::new(store);
+            let wake = todo_wake.clone();
+            tokio::spawn(async move {
+                if let Err(e) = todo_server::run(store, wake).await {
+                    tracing::error!(error = %e, "in-agent todo socket exited with error");
+                }
+            });
+        }
+        Err(e) => {
+            tracing::error!(error = ?e, "open todos db failed — in-agent todo socket disabled");
+        }
+    }
     if matches!(initial, LoginState::NeedsLogin) {
         login::wait_for_login(&claude_dir, login_state.clone(), &bus, poll_ms).await;
     } else {
@@ -441,6 +481,7 @@ async fn serve_main<S: Surface>(socket: &Path, poll_ms: u64) -> Result<()> {
         bus,
         stats,
         &files,
+        todo_wake,
     )
     .await
 }
@@ -462,6 +503,7 @@ async fn serve_loop<S: Surface>(
     bus: Bus,
     stats: Option<TurnStats>,
     files: &turn::TurnFiles,
+    todo_wake: Arc<tokio::sync::Notify>,
 ) -> Result<()> {
     tracing::info!(socket = %socket.display(), "harness serve");
     S::requeue_inflight(socket).await;
@@ -476,8 +518,20 @@ async fn serve_loop<S: Surface>(
     loop {
         let next = match self_continue.take() {
             Some(msg) => msg,
-            None => match S::recv_next(socket).await {
+            None => match {
+                // Idle wait: race the broker long-poll against a local
+                // todo signal so an in-container producer's upsert drives a
+                // turn without any broker round-trip. `biased` polls the
+                // broker recv first, so a genuinely-ready inbox message is
+                // never dropped in favour of the todo wake.
+                tokio::select! {
+                    biased;
+                    o = S::recv_next(socket) => o,
+                    () = todo_wake.notified() => RecvOutcome::LocalTodo,
+                }
+            } {
                 RecvOutcome::Message(first) => first,
+                RecvOutcome::LocalTodo => synthetic_todo_message(),
                 RecvOutcome::Empty => {
                     // Idle: no message this poll. Service a queued operator
                     // `/compact` here so it runs even when no turn is driving

hive-agent/src/todos.rs

@@ -33,7 +33,6 @@ CREATE TABLE IF NOT EXISTS todos (
     subsystem_key TEXT,
     summary       TEXT    NOT NULL,
     source        TEXT,
-    created_at    INTEGER NOT NULL,
     updated_at    INTEGER NOT NULL
 );
 -- (subsystem, subsystem_key) is the upsert/dedup key. A NULL key never
@@ -57,7 +56,6 @@ pub struct Todo {
     pub summary: String,
     /// Optional free-text provenance (e.g. the room name / task label).
     pub source: Option<String>,
-    pub created_at: i64,
     pub updated_at: i64,
 }
 
@@ -74,8 +72,8 @@ impl Todos {
     ///
     /// Propagates sqlite open / schema-apply failures.
     pub fn open(path: &Path) -> Result<Self> {
-        let conn = Connection::open(path)
+        let conn =
-            .with_context(|| format!("open todos db {}", path.display()))?;
+            Connection::open(path).with_context(|| format!("open todos db {}", path.display()))?;
         conn.execute_batch(SCHEMA).context("apply todos schema")?;
         Ok(Self {
             conn: Mutex::new(conn),
@@ -131,8 +129,8 @@ impl Todos {
         }
         conn.execute(
             "INSERT INTO todos \
-               (subsystem, subsystem_key, summary, source, created_at, updated_at) \
+               (subsystem, subsystem_key, summary, source, updated_at) \
-             VALUES (?1, ?2, ?3, ?4, ?5, ?5)",
+             VALUES (?1, ?2, ?3, ?4, ?5)",
             params![subsystem, key, summary, source, now],
         )?;
         Ok((conn.last_insert_rowid(), true))
@@ -173,10 +171,7 @@ impl Todos {
     /// Panics if the connection mutex is poisoned.
     pub fn clear_subsystem(&self, subsystem: &str) -> Result<usize> {
         let conn = self.conn.lock().unwrap();
-        let n = conn.execute(
+        let n = conn.execute("DELETE FROM todos WHERE subsystem = ?1", params![subsystem])?;
-            "DELETE FROM todos WHERE subsystem = ?1",
-            params![subsystem],
-        )?;
         Ok(n)
     }
 
@@ -209,7 +204,7 @@ impl Todos {
     pub fn list(&self, subsystem: Option<&str>) -> Result<Vec<Todo>> {
         let conn = self.conn.lock().unwrap();
         let mut stmt = conn.prepare(
-            "SELECT id, subsystem, subsystem_key, summary, source, created_at, updated_at \
+            "SELECT id, subsystem, subsystem_key, summary, source, updated_at \
              FROM todos \
              WHERE (?1 IS NULL OR subsystem = ?1) \
              ORDER BY updated_at DESC, id DESC",
@@ -222,8 +217,7 @@ impl Todos {
                     subsystem_key: row.get(2)?,
                     summary: row.get(3)?,
                     source: row.get(4)?,
-                    created_at: row.get(5)?,
+                    updated_at: row.get(5)?,
-                    updated_at: row.get(6)?,
                 })
             })?
             .collect::<rusqlite::Result<Vec<_>>>()?;
@@ -247,14 +241,20 @@ mod tests {
     #[test]
     fn keyed_upsert_dedups_and_reports_changed() {
         let (_dir, s) = store();
-        let (id1, changed1) = s.upsert("matrix", Some("!room:x"), "1 unread", None).unwrap();
+        let (id1, changed1) = s
+            .upsert("matrix", Some("!room:x"), "1 unread", None)
+            .unwrap();
         assert!(changed1, "first push is new → changed");
         // Same key + same summary → no-op, not changed (must not re-wake).
-        let (id2, changed2) = s.upsert("matrix", Some("!room:x"), "1 unread", None).unwrap();
+        let (id2, changed2) = s
+            .upsert("matrix", Some("!room:x"), "1 unread", None)
+            .unwrap();
         assert_eq!(id1, id2, "keyed upsert updates in place, same row");
         assert!(!changed2, "identical re-push is a no-op");
         // Same key, new summary → updates, changed.
-        let (id3, changed3) = s.upsert("matrix", Some("!room:x"), "3 unread", None).unwrap();
+        let (id3, changed3) = s
+            .upsert("matrix", Some("!room:x"), "3 unread", None)
+            .unwrap();
         assert_eq!(id1, id3);
         assert!(changed3);
         assert_eq!(s.list(Some("matrix")).unwrap().len(), 1);