From 86d16efa075754acd4f53355e6bf2312e41806c0 Mon Sep 17 00:00:00 2001 From: damocles Date: Mon, 20 Jul 2026 22:47:29 +0200 Subject: [PATCH] feat(#2569): drive a turn on a local todo signal (serve-loop select) --- hive-agent-sock/src/lib.rs | 2 +- hive-agent/src/main.rs | 56 +++++++++++++++++++++++++++++++++++++- hive-agent/src/todos.rs | 32 +++++++++++----------- 3 files changed, 72 insertions(+), 18 deletions(-) diff --git a/hive-agent-sock/src/lib.rs b/hive-agent-sock/src/lib.rs index dfe90539..dc7bd7a8 100644 --- a/hive-agent-sock/src/lib.rs +++ b/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). //! diff --git a/hive-agent/src/main.rs b/hive-agent/src/main.rs index 8f5374da..4c241696 100644 --- a/hive-agent/src/main.rs +++ b/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(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(socket: &Path, poll_ms: u64) -> Result<()> { bus, stats, &files, + todo_wake, ) .await } @@ -462,6 +503,7 @@ async fn serve_loop( bus: Bus, stats: Option, files: &turn::TurnFiles, + todo_wake: Arc, ) -> Result<()> { tracing::info!(socket = %socket.display(), "harness serve"); S::requeue_inflight(socket).await; @@ -476,8 +518,20 @@ async fn serve_loop( 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 diff --git a/hive-agent/src/todos.rs b/hive-agent/src/todos.rs index d95a6cb3..19af3142 100644 --- a/hive-agent/src/todos.rs +++ b/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, - 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 { - let conn = Connection::open(path) - .with_context(|| format!("open todos db {}", path.display()))?; + let conn = + 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) \ - VALUES (?1, ?2, ?3, ?4, ?5, ?5)", + (subsystem, subsystem_key, summary, source, updated_at) \ + 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 { let conn = self.conn.lock().unwrap(); - let n = conn.execute( - "DELETE FROM todos WHERE subsystem = ?1", - params![subsystem], - )?; + let n = conn.execute("DELETE FROM todos WHERE subsystem = ?1", params![subsystem])?; Ok(n) } @@ -209,7 +204,7 @@ impl Todos { pub fn list(&self, subsystem: Option<&str>) -> Result> { 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(6)?, + updated_at: row.get(5)?, }) })? .collect::>>()?; @@ -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);