diff --git a/hive-ag3nt/src/bash_tasks.rs b/hive-ag3nt/src/bash_tasks.rs deleted file mode 100644 index ce026f32..00000000 --- a/hive-ag3nt/src/bash_tasks.rs +++ /dev/null @@ -1,78 +0,0 @@ -//! Read-only view of bash task state files for `get_loose_ends`. -//! -//! The task runner and MCP tools live in `hive-bash-mcp`. This module -//! only reads the JSON task files that the daemon writes — no subprocess -//! logic, no wake signals, no dep on the hive-bash-mcp crate. -//! -//! Task files live under `harness_dir/bash-tasks/.json`. The JSON -//! schema is owned by `hive-bash-mcp::protocol::TaskFile`; this module -//! duplicates only the fields the harness needs for `get_loose_ends`. - -use std::path::PathBuf; - -use serde::{Deserialize, Serialize}; - -/// Lifecycle state of a bash task (mirrors `hive_bash_mcp::protocol::TaskStatus`). -#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] -#[serde(rename_all = "snake_case")] -pub enum TaskStatus { - Pending, - Running, - Done, - TimedOut, - Interrupted, -} - -/// Minimal task metadata (mirrors `hive_bash_mcp::protocol::TaskFile`). -#[derive(Debug, Clone, Deserialize)] -pub struct TaskFile { - pub id: String, - pub cmd: String, - pub status: TaskStatus, - pub created_at: i64, -} - -fn tasks_dir() -> PathBuf { - let base = if let Some(p) = std::env::var_os("HYPERHIVE_HARNESS_DIR") { - PathBuf::from(p) - } else { - let state = std::env::var("HYPERHIVE_STATE_DIR").unwrap_or_default(); - let state_path = PathBuf::from(&state); - state_path - .parent() - .map(|p| p.join("harness")) - .unwrap_or_else(|| PathBuf::from(state)) - }; - base.join("bash-tasks") -} - -/// Read a task file by ID. Returns `None` if missing or unparseable. -#[must_use] -pub fn read_task(id: &str) -> Option { - let path = tasks_dir().join(format!("{id}.json")); - let s = std::fs::read_to_string(path).ok()?; - serde_json::from_str(&s).ok() -} - -/// Return all tasks currently in `Pending` or `Running` state. -#[must_use] -pub fn active_tasks() -> Vec { - let Ok(rd) = std::fs::read_dir(tasks_dir()) else { - return Vec::new(); - }; - let mut out = Vec::new(); - for entry in rd.flatten() { - let path = entry.path(); - if path.extension().and_then(|e| e.to_str()) != Some("json") { - continue; - } - let Some(id) = path.file_stem().and_then(|s| s.to_str()).map(str::to_owned) else { - continue; - }; - let Some(task) = read_task(&id) else { continue }; - if matches!(task.status, TaskStatus::Pending | TaskStatus::Running) { - out.push(task); - } - } - out -} diff --git a/hive-ag3nt/src/lib.rs b/hive-ag3nt/src/lib.rs index 204e5917..801cc003 100644 --- a/hive-ag3nt/src/lib.rs +++ b/hive-ag3nt/src/lib.rs @@ -1,7 +1,6 @@ //! Shared in-container harness code used by both `hive-ag3nt` (agent) and //! `hive-m1nd` (manager) binaries. -pub mod bash_tasks; pub mod client; pub mod events; pub mod forge_notify; @@ -9,6 +8,7 @@ pub mod identity; pub mod login; pub mod login_session; pub mod mcp; +pub mod mcp_loose_ends; pub mod paths; pub mod plugins; pub mod prompt; diff --git a/hive-ag3nt/src/mcp.rs b/hive-ag3nt/src/mcp.rs index 1de441ba..2ff5c7d9 100644 --- a/hive-ag3nt/src/mcp.rs +++ b/hive-ag3nt/src/mcp.rs @@ -684,8 +684,8 @@ impl AgentServer { description = "List loose ends pending against this agent: unanswered questions \ where you are the asker (waiting on someone) or the target (someone's waiting on \ you), pending reminders you scheduled, plus — for the manager only — pending \ - approvals you submitted that the operator hasn't acted on yet. Also lists any \ - local bash tasks still in pending or running state. Cheap sweep, no args. Useful \ + approvals you submitted that the operator hasn't acted on yet. Also lists active \ + local tasks published by external MCP daemons (e.g. running bash tasks). Cheap sweep, no args. Useful \ at turn start to remember what you owe / what's owed to you without scrolling \ inbox history. Output is a short bulleted list with ids, ages in seconds, and \ the relevant context. Each `question` or `reminder` row can be cancelled by \ @@ -737,19 +737,15 @@ impl AgentServer { } } let mut out = annotate_retries(render_loose_ends(&loose_ends), retries); - // Append any local bash tasks still in pending/running state so - // the agent sees all outstanding work in one call. - let active = crate::bash_tasks::active_tasks(); - if !active.is_empty() { + // Append loose-end items published by external MCP daemons + // (e.g. active bash tasks from hive-bash-mcp). Generic — no + // per-MCP knowledge needed here. + let mcp_items = crate::mcp_loose_ends::collect(); + if !mcp_items.is_empty() { use std::fmt::Write as _; - let _ = write!(out, "\n\n{} active bash task(s):", active.len()); - for task in &active { - let age = crate::serve_common::now_unix() - task.created_at; - let _ = write!( - out, - "\n- `{}` status={:?}, cmd: `{}`, age {}s", - task.id, task.status, task.cmd, age - ); + let _ = write!(out, "\n\nlocal task(s):"); + for item in &mcp_items { + let _ = write!(out, "\n- {item}"); } } out diff --git a/hive-ag3nt/src/mcp_loose_ends.rs b/hive-ag3nt/src/mcp_loose_ends.rs new file mode 100644 index 00000000..e7e537e0 --- /dev/null +++ b/hive-ag3nt/src/mcp_loose_ends.rs @@ -0,0 +1,50 @@ +//! Generic scanner for MCP loose-end summary files. +//! +//! External MCP daemons (hive-bash-mcp, hive-matrix-mcp, etc.) write +//! JSON files to `$HYPERHIVE_HARNESS_DIR/mcp-loose-ends/.json`. +//! Each file contains a JSON array of plain-text summary strings. +//! +//! The harness reads all files in this directory in `get_loose_ends` to +//! surface active background work from any MCP without hardcoding +//! per-MCP knowledge here. + +use std::path::PathBuf; + +fn loose_ends_dir() -> PathBuf { + let base = if let Some(p) = std::env::var_os("HYPERHIVE_HARNESS_DIR") { + PathBuf::from(p) + } else { + let state = std::env::var("HYPERHIVE_STATE_DIR").unwrap_or_default(); + let state_path = PathBuf::from(&state); + state_path + .parent() + .map(|p| p.join("harness")) + .unwrap_or_else(|| PathBuf::from(state)) + }; + base.join("mcp-loose-ends") +} + +/// Collect all loose-end summary strings published by external MCP daemons. +/// Each string is a single line suitable for inclusion in `get_loose_ends` +/// output. Returns an empty vec if the directory doesn't exist or is empty. +#[must_use] +pub fn collect() -> Vec { + let Ok(rd) = std::fs::read_dir(loose_ends_dir()) else { + return Vec::new(); + }; + let mut out = Vec::new(); + for entry in rd.flatten() { + let path = entry.path(); + if path.extension().and_then(|e| e.to_str()) != Some("json") { + continue; + } + let Ok(content) = std::fs::read_to_string(&path) else { + continue; + }; + let Ok(items) = serde_json::from_str::>(&content) else { + continue; + }; + out.extend(items); + } + out +} diff --git a/hive-bash-mcp/src/paths.rs b/hive-bash-mcp/src/paths.rs index ce960f66..854bcc9c 100644 --- a/hive-bash-mcp/src/paths.rs +++ b/hive-bash-mcp/src/paths.rs @@ -48,6 +48,24 @@ pub fn hyperhive_socket() -> PathBuf { .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 { + let base = if let Some(p) = std::env::var_os("HYPERHIVE_HARNESS_DIR") { + PathBuf::from(p) + } else { + let state = std::env::var("HYPERHIVE_STATE_DIR").unwrap_or_default(); + let state_path = PathBuf::from(&state); + state_path + .parent() + .map(|p| p.join("harness")) + .unwrap_or_else(|| PathBuf::from(state)) + }; + base.join("mcp-loose-ends") +} + /// Full path for a task's JSON metadata file. #[must_use] pub fn task_json(id: &str) -> PathBuf { diff --git a/hive-bash-mcp/src/runner.rs b/hive-bash-mcp/src/runner.rs index 19f2dea4..f6a8ee14 100644 --- a/hive-bash-mcp/src/runner.rs +++ b/hive-bash-mcp/src/runner.rs @@ -84,6 +84,42 @@ pub fn read_task(id: &str) -> Option { serde_json::from_str(&s).ok() } +// --------------------------------------------------------------------------- +// Loose-ends file (generic MCP loose-ends protocol) +// --------------------------------------------------------------------------- + +/// 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"); + } +} + // --------------------------------------------------------------------------- // Public API used by daemon dispatch // --------------------------------------------------------------------------- @@ -110,6 +146,7 @@ pub fn submit_task(cmd: String, timeout_secs: Option) -> Result { stderr_tail: None, }; write_task(&task)?; + refresh_loose_ends(); Ok(id) } @@ -216,6 +253,7 @@ 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; } } @@ -266,6 +304,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(); let out_path = paths::task_out(&id); let err_path = paths::task_err(&id); @@ -299,6 +338,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 done state failed"); } + refresh_loose_ends(); let summary = if timed_out { format!("timed out after {}s", task.timeout_secs)