feat(#2626): migrate forge_notify to the in-agent todo socket (per-thread todos, not wakes)

hive-agent/src/forge_notify.rs

@@ -64,6 +64,8 @@ const NEW_ITEM_TOLERANCE_SECS: i64 = 120;
 /// configured. Otherwise loops forever, polling every
 /// `POLL_INTERVAL_SECS` seconds. Errors are never fatal.
 ///
+/// `socket` is the harness's in-agent todo socket (`HIVE_AGENT_SOCKET`):
+/// each forge notification is pushed as an `upsert_todo`, not a direct wake.
 pub async fn run(socket: PathBuf) {
     let forge_url = match std::env::var("HIVE_FORGE_URL") {
         Ok(u) if !u.is_empty() => u,
@@ -1064,33 +1066,40 @@ async fn poll_once(
             continue;
         };
 
-        let req = hive_core_agent_sock::Request::Wake {
+        // Upsert a *todo* (loose-ends v2) on the harness's in-agent socket,
-            from: "forge".to_owned(),
+        // keyed by the forge thread id, instead of firing a direct wake. A
-            body,
+        // new/changed summary makes the harness signal its turn loop; the
+        // agent clears the todo (`mark_todo_done`) once it has handled the
+        // thread. Re-scanning the same thread is an idempotent no-op.
+        let req = hive_agent_sock::Request::UpsertTodo {
+            subsystem: "forge".to_owned(),
+            key: Some(id.to_string()),
+            summary: body,
+            source: None,
         };
-        let deliver_result =
+        let deliver_result = crate::client::request::<_, hive_agent_sock::Response>(socket, &req)
-            crate::client::request::<_, hive_core_agent_sock::Response>(socket, &req)
+            .await
-                .await
+            .map(|_| ());
-                .map(|_| ());
         match deliver_result {
             Ok(()) => {
-                debug!(%id, "forge_notify: delivered");
+                debug!(%id, "forge_notify: todo upserted");
-                // Mark the thread read on forge immediately after a
+                // Mark the thread read on forge immediately after upserting
-                // successful broker delivery. The broker inbox is the
+                // the todo. The todo is the durable work item now (it stays
-                // durable work queue now (each row has its own ack
+                // in `get_loose_ends` until the agent marks it done), so the
-                // lifecycle), so the forge unread flag no longer needs to
+                // forge unread flag no longer needs to track agent
-                // track agent processing — clearing it on delivery keeps
+                // processing — clearing it keeps forge's unread set tiny by
-                // forge's unread set tiny by construction, so a container
+                // construction, so a container rebuild re-scan finds nothing
-                // rebuild re-scan finds nothing stale to re-deliver. The
+                // stale (and idempotent re-upserts wouldn't re-wake anyway).
-                // in-memory `delivered` entry below is only a within-process
+                // The in-memory `delivered` entry below is only a
-                // guard so a transient mark-read failure doesn't re-fire the
+                // within-process guard so a transient mark-read failure
-                // wake next tick; it is deliberately NOT persisted — forge's
+                // doesn't re-upsert next tick; it is deliberately NOT
-                // own read-state is the cross-rebuild source of truth.
+                // persisted — forge's own read-state is the cross-rebuild
+                // source of truth.
                 mark_read(forge, id).await;
                 delivered.insert(id, updated_at);
             }
             Err(e) => {
-                warn!(%id, error = ?e, "forge_notify: deliver failed — leaving unread");
+                warn!(%id, error = ?e, "forge_notify: todo upsert failed — leaving unread");
             }
         }
     }

hive-agent/src/main.rs

@@ -421,7 +421,13 @@ async fn serve_main<S: Surface>(socket: &Path, poll_ms: u64) -> Result<()> {
     for failure in plugins::install_configured().await {
         S::send_to_parent(socket, failure).await;
     }
-    tokio::spawn(crate::forge_notify::run(socket.to_path_buf()));
+    // forge_notify pushes forge notifications as todos on the in-agent
+    // socket (loose-ends v2), not direct wakes — so it dials
+    // `HIVE_AGENT_SOCKET`, not the host mcp.sock.
+    tokio::spawn(crate::forge_notify::run(
+        std::env::var_os("HIVE_AGENT_SOCKET")
+            .map_or_else(std::path::PathBuf::new, std::path::PathBuf::from),
+    ));
     // Agent-side cleanup of this agent's own harness artifacts (completed
     // bash-task files + verbose event rows). Runs here, not host-side in
     // hive-c0re, because the files are agent-owned — see `vacuum` module docs.