feat(#2628): migrate bash producer to the in-agent todo socket (keyed active + keyless done, not wakes)

Cargo.lock

@@ -1591,7 +1591,7 @@ name = "hive-bash-mcp"
 version = "0.1.0"
 dependencies = [
  "anyhow",
- "hive-core-agent-sock",
+ "hive-agent-sock",
  "hive-sh4re",
  "libc",
  "rmcp",

hive-bash-mcp/Cargo.toml

@@ -8,7 +8,7 @@ workspace = true
 
 [dependencies]
 anyhow.workspace = true
-hive-core-agent-sock.workspace = true
+hive-agent-sock.workspace = true
 hive-sh4re.workspace = true
 libc.workspace = true
 rmcp.workspace = true

hive-bash-mcp/src/main.rs

@@ -1,7 +1,8 @@
 //! `hive-bash-daemon` binary — long-running per-agent bash task runner.
 //! Spawns `sh -c` subprocesses, monitors completion, writes task state
-//! files, and fires hyperhive wake signals. Listens on a unix socket
+//! files, and surfaces task state to the agent as todos on the harness's
-//! for tool-call requests from the `hive-bash-mcp` stdio bridge.
+//! in-agent socket. Listens on a unix socket for tool-call requests from
+//! the `hive-bash-mcp` stdio bridge.
 
 use anyhow::Result;
 
@@ -15,17 +16,17 @@ async fn main() -> Result<()> {
         .init();
 
     let socket_path = hive_bash_mcp::paths::daemon_socket();
-    let wake_socket = hive_bash_mcp::paths::hyperhive_socket();
+    let todo_socket = hive_bash_mcp::paths::agent_socket();
 
     tracing::info!(
         socket = %socket_path.display(),
-        wake = %wake_socket.display(),
+        todo = %todo_socket.display(),
         "hive-bash-daemon starting"
     );
 
     // Start the background runner loop — scans for pending tasks and
-    // spawns them, sends wake signals on completion.
+    // spawns them, pushing todos to the harness on task transitions.
-    hive_bash_mcp::runner::spawn_loop(wake_socket);
+    hive_bash_mcp::runner::spawn_loop(todo_socket);
 
     // Serve the unix socket forever.
     hive_bash_mcp::socket::serve(&socket_path).await

hive-bash-mcp/src/paths.rs

@@ -43,21 +43,17 @@ pub fn turn_stats_db() -> PathBuf {
     harness_dir().join("hyperhive-turn-stats.sqlite")
 }
 
-/// Hyperhive control socket — the daemon writes wake signals here so
+/// The harness-served in-agent todo socket (loose-ends v2). The runner
-/// the harness drives a new claude turn on bash task completion.
+/// pushes bash-task todos here — `upsert_todo` while a task is active,
-/// Mirrors the path used by `forge_notify` and `hive-matrix-mcp`.
+/// a keyless `done` todo on completion — instead of firing a direct
+/// c0re wake. Override via `HIVE_AGENT_SOCKET`; mirrors the path the
+/// harness binds (`agent-service.nix`) and the matrix producer dials.
 #[must_use]
-pub fn hyperhive_socket() -> PathBuf {
+pub fn agent_socket() -> PathBuf {
-    std::env::var_os("HIVE_CONTROL_SOCKET")
+    std::env::var_os("HIVE_AGENT_SOCKET").map_or_else(
-        .map_or_else(|| PathBuf::from("/run/hive/mcp.sock"), PathBuf::from)
+        || PathBuf::from(hive_agent_sock::DEFAULT_AGENT_SOCKET),
-}
+        PathBuf::from,
-
+    )
-/// Directory where MCP daemons write loose-end summary files for the harness.
-/// Each daemon writes `<name>.json` here; the harness scans the dir in
-/// `get_loose_ends` to surface active work from all MCPs generically.
-#[must_use]
-pub fn mcp_loose_ends_dir() -> PathBuf {
-    hive_sh4re::paths::mcp_loose_ends_dir()
 }
 
 /// Full path for a task's JSON metadata file.

hive-bash-mcp/src/runner.rs

@@ -1,15 +1,21 @@
 //! Bash subprocess runner: spawns `bash -c <cmd>` tasks, writes status
-//! files under `harness/bash-tasks/`, and fires hyperhive wake signals
+//! files under `harness/bash-tasks/`, and surfaces task state to the agent
-//! on completion.
+//! as *todos* on the harness's in-agent socket (loose-ends v2).
 //!
 //! Files under `tasks_dir()`:
 //! - `<id>.json`  — task metadata + status (pending → running → done)
 //! - `<id>.out`   — captured stdout (streamed while running)
 //! - `<id>.err`   — captured stderr (streamed while running)
 //!
+//! Todos (loose-ends v2): a keyed todo (`key = task id`) is upserted while
+//! a task is active and cleared when it finishes; a keyless one-off `done`
+//! todo then carries the completion summary, which signals the harness turn
+//! loop the same way the old completion wake did. The agent clears a `done`
+//! todo with `mark_todo_done` once it has read the output.
+//!
 //! Tasks with status `running` on daemon boot are marked `interrupted`
-//! (the process died with the previous daemon). A best-effort wake is
+//! (the process died with the previous daemon). A best-effort `done` todo
-//! still sent so the agent is not silently blocked.
+//! is still pushed so the agent is not silently blocked.
 //!
 //! The runner kills the child process on timeout — `tokio::process::Child::drop()`
 //! does not kill children, so we explicitly call `child.kill().await`.
@@ -21,6 +27,7 @@ use std::sync::{Arc, Mutex, OnceLock};
 use std::time::{Duration, SystemTime, UNIX_EPOCH};
 
 use anyhow::{Result, bail};
+use hive_agent_sock::Request as TodoReq;
 use tokio::io::AsyncWriteExt as _;
 use tokio::sync::Notify;
 
@@ -157,41 +164,97 @@ pub fn read_task(id: &str) -> Option<TaskFile> {
 }
 
 // ---------------------------------------------------------------------------
-// Loose-ends file (generic MCP loose-ends protocol)
+// Todo delivery (loose-ends v2)
 // ---------------------------------------------------------------------------
+//
+// Bash-task state is surfaced to the agent as todos on the harness's
+// in-agent socket (`HIVE_AGENT_SOCKET`), not as direct c0re wakes: a keyed
+// todo (`key = task id`) is upserted while the task is active and cleared
+// when it finishes, and a keyless one-off `done` todo carries the
+// completion summary. A fresh keyless row always signals the harness turn
+// loop, so it drives a turn exactly like the old wake did.
 
-/// Rewrite `mcp-loose-ends/bash.json` with a summary of all currently
+/// Send one todo request to the harness in-agent socket. Best-effort: any
-/// active (Pending or Running) tasks. The harness scans this directory
+/// connect / write error is logged and swallowed — the harness self-heals
-/// generically in `get_loose_ends` — no bash-specific code needed there.
+/// on the next task transition, and a standalone daemon without the socket
-///
+/// simply has nowhere to push.
-/// File format: a JSON array of plain-text summary strings, one per
+async fn send_todo(socket: &Path, req: &TodoReq) {
-/// loose-end item. The harness includes them verbatim in the output.
+    use tokio::io::{AsyncBufReadExt as _, BufReader};
-/// Atomic write (tmp + rename) so the harness never reads a partial file.
+    use tokio::net::UnixStream;
-fn refresh_loose_ends() {
+    let line = match serde_json::to_string(req) {
-    let active = active_tasks();
+        Ok(mut s) => {
-    let dir = paths::mcp_loose_ends_dir();
+            s.push('\n');
-    if let Err(e) = std::fs::create_dir_all(&dir) {
+            s
-        tracing::warn!(error = ?e, "bash_runner: create mcp-loose-ends dir failed");
+        }
-        return;
+        Err(e) => {
-    }
+            tracing::warn!(error = ?e, "bash_runner: serialise todo failed");
-    let items: Vec<String> = active
+            return;
-        .iter()
+        }
-        .map(|t| {
+    };
-            let age = now_unix() - t.created_at;
+    match UnixStream::connect(socket).await {
-            format!(
+        Ok(stream) => {
-                "bash task `{}` status={:?}, cmd: `{}`, age {}s",
+            let (read, mut write) = stream.into_split();
-                t.id, t.status, t.cmd, age
+            if write.write_all(line.as_bytes()).await.is_err() {
-            )
+                tracing::warn!("bash_runner: write todo failed");
-        })
+                return;
-        .collect();
+            }
-    let dest = dir.join("bash.json");
+            let _ = write.shutdown().await;
-    let tmp = dest.with_extension("json.tmp");
+            // Drain the response so the server doesn't get ECONNRESET.
-    let json = serde_json::to_string(&items).unwrap_or_else(|_| "[]".to_owned());
+            let mut resp = String::new();
-    if let Err(e) = std::fs::write(&tmp, &json).and_then(|()| std::fs::rename(&tmp, &dest)) {
+            let _ = BufReader::new(read).read_line(&mut resp).await;
-        tracing::warn!(error = ?e, "bash_runner: write mcp-loose-ends/bash.json failed");
+        }
+        Err(e) => {
+            tracing::warn!(error = ?e, socket = %socket.display(), "bash_runner: connect todo socket failed");
+        }
     }
 }
 
+/// Upsert the keyed "active" todo for a running task (surfaces it in
+/// `get_loose_ends`). The summary is stable across the task's lifetime, so
+/// re-pushing it is an idempotent no-op that never re-wakes.
+async fn upsert_active_todo(socket: &Path, id: &str, cmd: &str) {
+    send_todo(
+        socket,
+        &TodoReq::UpsertTodo {
+            subsystem: "bash".to_owned(),
+            key: Some(id.to_owned()),
+            summary: format!("bash task `{id}` running: `{cmd}`"),
+            source: None,
+        },
+    )
+    .await;
+}
+
+/// Clear a task's keyed "active" todo once it has reached a terminal state.
+async fn clear_active_todo(socket: &Path, id: &str) {
+    send_todo(
+        socket,
+        &TodoReq::ClearTodo {
+            subsystem: "bash".to_owned(),
+            key: Some(id.to_owned()),
+            all: false,
+        },
+    )
+    .await;
+}
+
+/// Push a keyless one-off `done` todo carrying the completion `summary` —
+/// the loose-ends-v2 replacement for the old completion wake. A fresh
+/// keyless row always signals the turn loop, so the agent is driven a turn
+/// to read the output (and clears the todo with `mark_todo_done`).
+async fn push_done_todo(socket: &Path, summary: String) {
+    send_todo(
+        socket,
+        &TodoReq::UpsertTodo {
+            subsystem: "bash".to_owned(),
+            key: None,
+            summary,
+            source: None,
+        },
+    )
+    .await;
+}
+
 // ---------------------------------------------------------------------------
 // Public API used by daemon dispatch
 // ---------------------------------------------------------------------------
@@ -265,7 +328,10 @@ pub fn submit_task(cmd: String, timeout_secs: Option<u64>, name: Option<String>)
         stderr_tail: None,
     };
     write_task(&task)?;
-    refresh_loose_ends();
+    // No todo yet: the keyed "active" todo is upserted when the runner
+    // loop flips the task to Running (`run_task`), which has the in-agent
+    // socket handle. Surfacing a Pending task would only race the ~200ms
+    // until it starts and double-wake (pending→running).
     Ok(id)
 }
 
@@ -385,7 +451,9 @@ pub fn kill_task(id: &str, force: bool) -> (bool, bool) {
         task.status = TaskStatus::Killed;
         task.completed_at = Some(now_unix());
         let _ = write_task(&task);
-        refresh_loose_ends();
+        // A Pending task never reached Running, so it has no keyed "active"
+        // todo to clear and (like the old code) fires no completion signal —
+        // the caller that killed it already knows.
         return (true, false);
     }
     (false, false)
@@ -395,9 +463,9 @@ pub fn kill_task(id: &str, force: bool) -> (bool, bool) {
 // Runner background loop
 // ---------------------------------------------------------------------------
 
-/// Spawn the background runner loop as a detached tokio task. `socket`
+/// Spawn the background runner loop as a detached tokio task. `socket` is
-/// is the path to the per-agent broker socket used to deliver completion
+/// the harness's in-agent todo socket (`HIVE_AGENT_SOCKET`), where task
-/// wake signals. Call once at daemon startup.
+/// transitions are pushed as todos. Call once at daemon startup.
 pub fn spawn_loop(socket: PathBuf) {
     tokio::spawn(async move {
         run_loop(socket).await;
@@ -443,8 +511,12 @@ async fn mark_interrupted(socket: &Path) {
         if let Err(e) = write_task(&task) {
             tracing::warn!(id = %id, error = ?e, "bash_runner: write interrupted state failed");
         }
-        refresh_loose_ends();
+        clear_active_todo(socket, &id).await;
-        send_wake(socket, &id, "interrupted (daemon restarted)", None).await;
+        push_done_todo(
+            socket,
+            done_summary(&id, "interrupted (daemon restarted)", None),
+        )
+        .await;
     }
 }
 
@@ -494,7 +566,7 @@ async fn run_task(mut task: TaskFile, socket: &Path) {
     if let Err(e) = write_task(&task) {
         tracing::warn!(id = %id, error = ?e, "bash_runner: write running state failed");
     }
-    refresh_loose_ends();
+    upsert_active_todo(socket, &id, &task.cmd).await;
     // Best-effort: tally the normalised command head for the /stats
     // "favorite tools" view. Counted once per execution, regardless of
     // exit status. Never fails the task.
@@ -562,25 +634,31 @@ async fn run_task(mut task: TaskFile, socket: &Path) {
     if let Err(e) = write_task(&task) {
         tracing::warn!(id = %id, error = ?e, "bash_runner: write done state failed");
     }
-    refresh_loose_ends();
+    // Always retire the keyed "active" todo — the task has finished
+    // regardless of whether its completion also gets a `done` todo below.
+    clear_active_todo(socket, &id).await;
 
-    // Skip the wake if a `status`/`run` inline wait already handed this
+    // Skip the `done` todo if a `status`/`run` inline wait already handed
-    // exact terminal result to the caller in a tool response — see
+    // this exact terminal result to the caller in a tool response — see
     // `wait_for_task` / `observe_terminal`. Narrow race: an inline waiter
     // that polls in the few hundred ms right around this point may lose the
-    // race and still get a wake alongside its inline result; best-effort,
+    // race and still get a `done` todo alongside its inline result;
-    // same tolerance as the rest of this daemon's delivery guarantees.
+    // best-effort, same tolerance as the rest of this daemon's guarantees.
     if take_wake_suppressed(&id) {
-        tracing::debug!(id = %id, "bash_runner: wake suppressed (already observed via status)");
+        tracing::debug!(id = %id, "bash_runner: done todo suppressed (already observed via status)");
         return;
     }
 
-    // Pass only whether each stream produced (trimmed) output — `send_wake`
+    // Pass only whether each stream produced (trimmed) output — the `done`
-    // emits a `Read(<path>)` pointer to the captured `.out`/`.err` files
+    // todo carries a `Read(<path>)` pointer to the captured `.out`/`.err`
-    // rather than inlining the tail into the wake body.
+    // files rather than inlining the tail into the summary.
     let has_stdout = !stdout_tail.as_deref().unwrap_or("").trim().is_empty();
     let has_stderr = !stderr_tail.as_deref().unwrap_or("").trim().is_empty();
-    send_wake(socket, &id, &summary, Some((has_stdout, has_stderr))).await;
+    push_done_todo(
+        socket,
+        done_summary(&id, &summary, Some((has_stdout, has_stderr))),
+    )
+    .await;
 }
 
 /// Run `bash -c cmd` in its own process group, streaming output to files.
@@ -726,24 +804,20 @@ fn tail_file(path: &Path, max_bytes: usize) -> Option<String> {
 }
 
 // ---------------------------------------------------------------------------
-// Wake delivery
+// Completion summary
 // ---------------------------------------------------------------------------
 
-pub(crate) async fn send_wake(
+/// Build the completion-summary line carried by a `done` todo: the status
-    socket: &Path,
+/// `summary` plus `Read(<path>)` pointers to any captured output.
-    id: &str,
+///
-    summary: &str,
+/// `output` is `(has_stdout, has_stderr)` — whether the task produced
-    // `(has_stdout, has_stderr)` — whether the task produced non-empty
+/// non-empty stdout / stderr; `None` for the interrupted path, which has no
-    // stdout / stderr. `None` for the interrupted path (no captured output
+/// captured output. Deliberately NOT the output text: the todo gives the
-    // to point at). Deliberately NOT the output text: the wake gives the
+/// agent the status + a `Read(<path>)` pointer to the captured files rather
-    // agent the status + a `Read(<path>)` pointer to the captured files
+/// than inlining the tail, which would flood the agent's context on every
-    // rather than inlining the tail, which would flood the agent's context
+/// task completion.
-    // on every task completion.
+fn done_summary(id: &str, summary: &str, output: Option<(bool, bool)>) -> String {
-    output: Option<(bool, bool)>,
-) {
     use std::fmt::Write as _;
-    use tokio::io::{AsyncBufReadExt as _, BufReader};
-    use tokio::net::UnixStream;
     let mut body = format!("bash task `{id}` finished: {summary}");
     if let Some((has_stdout, has_stderr)) = output
         && (has_stdout || has_stderr)
@@ -760,38 +834,7 @@ pub(crate) async fn send_wake(
             );
         }
     }
-    let req = hive_core_agent_sock::Request::Wake {
+    body
-        from: format!("bash-task-{id}"),
-        body,
-    };
-
-    match UnixStream::connect(socket).await {
-        Ok(stream) => {
-            let (read, mut write) = stream.into_split();
-            let line = match serde_json::to_string(&req) {
-                Ok(mut s) => {
-                    s.push('\n');
-                    s
-                }
-                Err(e) => {
-                    tracing::warn!(id = %id, error = ?e, "bash_runner: serialise wake failed");
-                    return;
-                }
-            };
-            if write.write_all(line.as_bytes()).await.is_err() {
-                tracing::warn!(id = %id, "bash_runner: write wake failed");
-                return;
-            }
-            let _ = write.shutdown().await;
-            // Drain the response so the server doesn't get ECONNRESET.
-            let mut resp = String::new();
-            let _ = BufReader::new(read).read_line(&mut resp).await;
-            tracing::info!(id = %id, "bash_runner: wake delivered");
-        }
-        Err(e) => {
-            tracing::warn!(id = %id, error = ?e, "bash_runner: connect wake socket failed");
-        }
-    }
 }
 
 #[cfg(test)]

nix/agent-modules/mcp.nix

@@ -175,7 +175,11 @@ in
       ];
       environment = {
         HIVE_BASH_SOCKET = "/run/hive-bash/socket";
-        HIVE_CONTROL_SOCKET = "/run/hive/mcp.sock";
+        # In-agent todo socket the harness serves (loose-ends v2): the
+        # runner pushes bash-task todos here (upsert while active, keyless
+        # 'done' on completion) instead of firing a c0re wake. Must match
+        # the harness's HIVE_AGENT_SOCKET (agent-service.nix).
+        HIVE_AGENT_SOCKET = "/run/hive-agent/${userName}/agent.sock";
         RUST_LOG = "info";
         # HYPERHIVE_HARNESS_DIR and HYPERHIVE_STATE_DIR are already
         # injected via systemd.globalEnvironment by the meta flake