fix: replace bash_tasks.rs with generic mcp_loose_ends scanner
- hive-ag3nt: remove bash_tasks.rs entirely; add mcp_loose_ends.rs that scans harness/mcp-loose-ends/*.json generically (no bash knowledge) - hive-bash-mcp: daemon writes mcp-loose-ends/bash.json on every task state transition (pending/running/done/interrupted/timed_out) - get_loose_ends: reads mcp_loose_ends::collect() instead of bash-specific code - implements the generic mechanism from #1162
This commit is contained in:
parent
8b559eceec
commit
8731474104
6 changed files with 119 additions and 93 deletions
|
|
@ -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/<id>.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<TaskFile> {
|
|
||||||
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<TaskFile> {
|
|
||||||
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
|
|
||||||
}
|
|
||||||
|
|
@ -1,7 +1,6 @@
|
||||||
//! Shared in-container harness code used by both `hive-ag3nt` (agent) and
|
//! Shared in-container harness code used by both `hive-ag3nt` (agent) and
|
||||||
//! `hive-m1nd` (manager) binaries.
|
//! `hive-m1nd` (manager) binaries.
|
||||||
|
|
||||||
pub mod bash_tasks;
|
|
||||||
pub mod client;
|
pub mod client;
|
||||||
pub mod events;
|
pub mod events;
|
||||||
pub mod forge_notify;
|
pub mod forge_notify;
|
||||||
|
|
@ -9,6 +8,7 @@ pub mod identity;
|
||||||
pub mod login;
|
pub mod login;
|
||||||
pub mod login_session;
|
pub mod login_session;
|
||||||
pub mod mcp;
|
pub mod mcp;
|
||||||
|
pub mod mcp_loose_ends;
|
||||||
pub mod paths;
|
pub mod paths;
|
||||||
pub mod plugins;
|
pub mod plugins;
|
||||||
pub mod prompt;
|
pub mod prompt;
|
||||||
|
|
|
||||||
|
|
@ -684,8 +684,8 @@ impl AgentServer {
|
||||||
description = "List loose ends pending against this agent: unanswered questions \
|
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 \
|
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 \
|
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 \
|
approvals you submitted that the operator hasn't acted on yet. Also lists active \
|
||||||
local bash tasks still in pending or running state. Cheap sweep, no args. Useful \
|
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 \
|
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 \
|
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 \
|
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);
|
let mut out = annotate_retries(render_loose_ends(&loose_ends), retries);
|
||||||
// Append any local bash tasks still in pending/running state so
|
// Append loose-end items published by external MCP daemons
|
||||||
// the agent sees all outstanding work in one call.
|
// (e.g. active bash tasks from hive-bash-mcp). Generic — no
|
||||||
let active = crate::bash_tasks::active_tasks();
|
// per-MCP knowledge needed here.
|
||||||
if !active.is_empty() {
|
let mcp_items = crate::mcp_loose_ends::collect();
|
||||||
|
if !mcp_items.is_empty() {
|
||||||
use std::fmt::Write as _;
|
use std::fmt::Write as _;
|
||||||
let _ = write!(out, "\n\n{} active bash task(s):", active.len());
|
let _ = write!(out, "\n\nlocal task(s):");
|
||||||
for task in &active {
|
for item in &mcp_items {
|
||||||
let age = crate::serve_common::now_unix() - task.created_at;
|
let _ = write!(out, "\n- {item}");
|
||||||
let _ = write!(
|
|
||||||
out,
|
|
||||||
"\n- `{}` status={:?}, cmd: `{}`, age {}s",
|
|
||||||
task.id, task.status, task.cmd, age
|
|
||||||
);
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
out
|
out
|
||||||
|
|
|
||||||
50
hive-ag3nt/src/mcp_loose_ends.rs
Normal file
50
hive-ag3nt/src/mcp_loose_ends.rs
Normal file
|
|
@ -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/<name>.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<String> {
|
||||||
|
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::<Vec<String>>(&content) else {
|
||||||
|
continue;
|
||||||
|
};
|
||||||
|
out.extend(items);
|
||||||
|
}
|
||||||
|
out
|
||||||
|
}
|
||||||
|
|
@ -48,6 +48,24 @@ pub fn hyperhive_socket() -> PathBuf {
|
||||||
.map_or_else(|| PathBuf::from("/run/hive/mcp.sock"), PathBuf::from)
|
.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 `<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 {
|
||||||
|
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.
|
/// Full path for a task's JSON metadata file.
|
||||||
#[must_use]
|
#[must_use]
|
||||||
pub fn task_json(id: &str) -> PathBuf {
|
pub fn task_json(id: &str) -> PathBuf {
|
||||||
|
|
|
||||||
|
|
@ -84,6 +84,42 @@ pub fn read_task(id: &str) -> Option<TaskFile> {
|
||||||
serde_json::from_str(&s).ok()
|
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<String> = 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
|
// Public API used by daemon dispatch
|
||||||
// ---------------------------------------------------------------------------
|
// ---------------------------------------------------------------------------
|
||||||
|
|
@ -110,6 +146,7 @@ pub fn submit_task(cmd: String, timeout_secs: Option<u64>) -> Result<String> {
|
||||||
stderr_tail: None,
|
stderr_tail: None,
|
||||||
};
|
};
|
||||||
write_task(&task)?;
|
write_task(&task)?;
|
||||||
|
refresh_loose_ends();
|
||||||
Ok(id)
|
Ok(id)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
@ -216,6 +253,7 @@ async fn mark_interrupted(socket: &Path) {
|
||||||
if let Err(e) = write_task(&task) {
|
if let Err(e) = write_task(&task) {
|
||||||
tracing::warn!(id = %id, error = ?e, "bash_runner: write interrupted state failed");
|
tracing::warn!(id = %id, error = ?e, "bash_runner: write interrupted state failed");
|
||||||
}
|
}
|
||||||
|
refresh_loose_ends();
|
||||||
send_wake(socket, &id, "interrupted (daemon restarted)", None).await;
|
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) {
|
if let Err(e) = write_task(&task) {
|
||||||
tracing::warn!(id = %id, error = ?e, "bash_runner: write running state failed");
|
tracing::warn!(id = %id, error = ?e, "bash_runner: write running state failed");
|
||||||
}
|
}
|
||||||
|
refresh_loose_ends();
|
||||||
|
|
||||||
let out_path = paths::task_out(&id);
|
let out_path = paths::task_out(&id);
|
||||||
let err_path = paths::task_err(&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) {
|
if let Err(e) = write_task(&task) {
|
||||||
tracing::warn!(id = %id, error = ?e, "bash_runner: write done state failed");
|
tracing::warn!(id = %id, error = ?e, "bash_runner: write done state failed");
|
||||||
}
|
}
|
||||||
|
refresh_loose_ends();
|
||||||
|
|
||||||
let summary = if timed_out {
|
let summary = if timed_out {
|
||||||
format!("timed out after {}s", task.timeout_secs)
|
format!("timed out after {}s", task.timeout_secs)
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue