diff --git a/Cargo.lock b/Cargo.lock index 0d5b1376..57531fd2 100644 --- a/Cargo.lock +++ b/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", diff --git a/hive-bash-mcp/Cargo.toml b/hive-bash-mcp/Cargo.toml index 440e73a1..06269257 100644 --- a/hive-bash-mcp/Cargo.toml +++ b/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 diff --git a/hive-bash-mcp/src/main.rs b/hive-bash-mcp/src/main.rs index 8cfbe8ff..d7c1e166 100644 --- a/hive-bash-mcp/src/main.rs +++ b/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 -//! for tool-call requests from the `hive-bash-mcp` stdio bridge. +//! files, and surfaces task state to the agent as todos on the harness's +//! 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. - hive_bash_mcp::runner::spawn_loop(wake_socket); + // spawns them, pushing todos to the harness on task transitions. + hive_bash_mcp::runner::spawn_loop(todo_socket); // Serve the unix socket forever. hive_bash_mcp::socket::serve(&socket_path).await diff --git a/hive-bash-mcp/src/paths.rs b/hive-bash-mcp/src/paths.rs index 90714e4b..478861fd 100644 --- a/hive-bash-mcp/src/paths.rs +++ b/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 drives a new claude turn on bash task completion. -/// Mirrors the path used by `forge_notify` and `hive-matrix-mcp`. +/// The harness-served in-agent todo socket (loose-ends v2). The runner +/// pushes bash-task todos here — `upsert_todo` while a task is active, +/// 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 { - std::env::var_os("HIVE_CONTROL_SOCKET") - .map_or_else(|| PathBuf::from("/run/hive/mcp.sock"), PathBuf::from) -} - -/// Directory where MCP daemons write loose-end summary files for the harness. -/// Each daemon writes `.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() +pub fn agent_socket() -> PathBuf { + std::env::var_os("HIVE_AGENT_SOCKET").map_or_else( + || PathBuf::from(hive_agent_sock::DEFAULT_AGENT_SOCKET), + PathBuf::from, + ) } /// Full path for a task's JSON metadata file. diff --git a/hive-bash-mcp/src/runner.rs b/hive-bash-mcp/src/runner.rs index bba8dcb6..f5f2d281 100644 --- a/hive-bash-mcp/src/runner.rs +++ b/hive-bash-mcp/src/runner.rs @@ -1,15 +1,21 @@ //! Bash subprocess runner: spawns `bash -c ` tasks, writes status -//! files under `harness/bash-tasks/`, and fires hyperhive wake signals -//! on completion. +//! files under `harness/bash-tasks/`, and surfaces task state to the agent +//! as *todos* on the harness's in-agent socket (loose-ends v2). //! //! Files under `tasks_dir()`: //! - `.json` — task metadata + status (pending → running → done) //! - `.out` — captured stdout (streamed while running) //! - `.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 -//! still sent so the agent is not silently blocked. +//! (the process died with the previous daemon). A best-effort `done` todo +//! 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 { } // --------------------------------------------------------------------------- -// 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 -/// active (Pending or Running) tasks. The harness scans this directory -/// generically in `get_loose_ends` — no bash-specific code needed there. -/// -/// File format: a JSON array of plain-text summary strings, one per -/// loose-end item. The harness includes them verbatim in the output. -/// Atomic write (tmp + rename) so the harness never reads a partial file. -fn refresh_loose_ends() { - let active = active_tasks(); - let dir = paths::mcp_loose_ends_dir(); - if let Err(e) = std::fs::create_dir_all(&dir) { - tracing::warn!(error = ?e, "bash_runner: create mcp-loose-ends dir failed"); - return; - } - let items: Vec = active - .iter() - .map(|t| { - let age = now_unix() - t.created_at; - format!( - "bash task `{}` status={:?}, cmd: `{}`, age {}s", - t.id, t.status, t.cmd, age - ) - }) - .collect(); - let dest = dir.join("bash.json"); - let tmp = dest.with_extension("json.tmp"); - let json = serde_json::to_string(&items).unwrap_or_else(|_| "[]".to_owned()); - if let Err(e) = std::fs::write(&tmp, &json).and_then(|()| std::fs::rename(&tmp, &dest)) { - tracing::warn!(error = ?e, "bash_runner: write mcp-loose-ends/bash.json failed"); +/// Send one todo request to the harness in-agent socket. Best-effort: any +/// connect / write error is logged and swallowed — the harness self-heals +/// on the next task transition, and a standalone daemon without the socket +/// simply has nowhere to push. +async fn send_todo(socket: &Path, req: &TodoReq) { + use tokio::io::{AsyncBufReadExt as _, BufReader}; + use tokio::net::UnixStream; + let line = match serde_json::to_string(req) { + Ok(mut s) => { + s.push('\n'); + s + } + Err(e) => { + tracing::warn!(error = ?e, "bash_runner: serialise todo failed"); + return; + } + }; + match UnixStream::connect(socket).await { + Ok(stream) => { + let (read, mut write) = stream.into_split(); + if write.write_all(line.as_bytes()).await.is_err() { + tracing::warn!("bash_runner: write todo 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; + } + 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, name: Option) 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` -/// is the path to the per-agent broker socket used to deliver completion -/// wake signals. Call once at daemon startup. +/// Spawn the background runner loop as a detached tokio task. `socket` is +/// the harness's in-agent todo socket (`HIVE_AGENT_SOCKET`), where task +/// 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(); - send_wake(socket, &id, "interrupted (daemon restarted)", None).await; + clear_active_todo(socket, &id).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 - // exact terminal result to the caller in a tool response — see + // Skip the `done` todo if a `status`/`run` inline wait already handed + // 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, - // same tolerance as the rest of this daemon's delivery guarantees. + // race and still get a `done` todo alongside its inline result; + // 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` - // emits a `Read()` pointer to the captured `.out`/`.err` files - // rather than inlining the tail into the wake body. + // Pass only whether each stream produced (trimmed) output — the `done` + // todo carries a `Read()` pointer to the captured `.out`/`.err` + // 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 { } // --------------------------------------------------------------------------- -// Wake delivery +// Completion summary // --------------------------------------------------------------------------- -pub(crate) async fn send_wake( - socket: &Path, - id: &str, - summary: &str, - // `(has_stdout, has_stderr)` — whether the task produced non-empty - // stdout / stderr. `None` for the interrupted path (no captured output - // to point at). Deliberately NOT the output text: the wake gives the - // agent the status + a `Read()` pointer to the captured files - // rather than inlining the tail, which would flood the agent's context - // on every task completion. - output: Option<(bool, bool)>, -) { +/// Build the completion-summary line carried by a `done` todo: the status +/// `summary` plus `Read()` pointers to any captured output. +/// +/// `output` is `(has_stdout, has_stderr)` — whether the task produced +/// non-empty stdout / stderr; `None` for the interrupted path, which has no +/// captured output. Deliberately NOT the output text: the todo gives the +/// agent the status + a `Read()` pointer to the captured files rather +/// than inlining the tail, which would flood the agent's context on every +/// task completion. +fn done_summary(id: &str, summary: &str, output: Option<(bool, bool)>) -> String { 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 { - 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"); - } - } + body } #[cfg(test)] diff --git a/nix/agent-modules/mcp.nix b/nix/agent-modules/mcp.nix index 6a6aaeca..07fab953 100644 --- a/nix/agent-modules/mcp.nix +++ b/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