diff --git a/Cargo.lock b/Cargo.lock index 57531fd2..0d5b1376 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1591,7 +1591,7 @@ name = "hive-bash-mcp" version = "0.1.0" dependencies = [ "anyhow", - "hive-agent-sock", + "hive-core-agent-sock", "hive-sh4re", "libc", "rmcp", diff --git a/docs/persistence.md b/docs/persistence.md index 10495ef4..006db829 100644 --- a/docs/persistence.md +++ b/docs/persistence.md @@ -233,12 +233,13 @@ Under `/var/lib/hyperhive/agents//`: (full captured output). `hive-c0re::bash_tasks_vacuum` runs hourly and deletes terminal task trios older than 48 hours; non-terminal (still-running) tasks are never deleted by vacuum. - - `hyperhive-todos.sqlite` — loose-ends-v2 todo store. In-container - MCP daemons (`hive-bash-mcp`, `hive-matrix-mcp`) and `forge_notify` - upsert keyed todos here over the harness's in-agent socket - (`HIVE_AGENT_SOCKET`); the harness merges them into `get_loose_ends` - output and clears a row on `mark_todo_done`. Replaced the old - file-based `mcp-loose-ends/` scanner. + - `mcp-loose-ends/` — JSON files published by external MCP daemons + (e.g. `hive-bash-mcp`, `hive-matrix-mcp`) listing their active + loose-end summary strings. Each file is `.json` containing + a JSON array of plain-text lines. Read by `get_loose_ends` to + surface active background work without hardcoding per-MCP + knowledge in the harness. Files are created/removed by the + external daemons themselves. - `hyperhive-events.sqlite` — turn-loop event log. - `hyperhive-turn-stats.sqlite` — per-turn timing stats. - `hyperhive-model` — single-line model name override file. diff --git a/docs/tools/bash.md b/docs/tools/bash.md index 1e802ac1..cac557af 100644 --- a/docs/tools/bash.md +++ b/docs/tools/bash.md @@ -13,24 +13,24 @@ invocation regardless of tool groups. Submit a shell command for background execution (runs via `bash`). Stdout and stderr stream to `harness/bash-tasks/.{out,err}`. -When the task completes (or times out, or the process errors), it -surfaces as a todo in the agent's loose-ends (via `get_loose_ends`), -carrying the exit code and a `Read()` pointer to the captured -output. Handle it on a future turn — unless `wait_seconds` already -delivered the terminal result inline, in which case no todo is created -(see `status` below). +When the task completes (or times out, or the process errors), the +harness fires a wake with `from: "bash-task-"`; the body contains +the exit code and last stdout lines. Handle the completion on a future +turn — unless `wait_seconds` already delivered the terminal result +inline, in which case the wake is suppressed (see `status` below). * `timeout_secs` — kill the task after N seconds and mark it `timed_out`. Omit for no timeout (runs until natural exit). * `wait_seconds` — inline poll before returning (capped at 30). When the task finishes within the window the full status is - returned immediately and no todo is created; when the window expires + returned immediately and no wake is fired; when the window expires the task keeps running and the normal `task started: id=` response is returned. **Defaults to 3** — pass `wait_seconds: 0` to disable inline waiting and always get the immediate response. * `name` — optional caller-chosen task id. When set it replaces the - auto-generated hex id, so it surfaces in `status()` lookups and - the loose-ends list — a memorable label instead of an opaque id. A name + auto-generated hex id, so it surfaces in the wake `from` + (`bash-task-`), in `status()` lookups, and in the + loose-ends list — a memorable label instead of an opaque id. A name is **reusable once its previous task has finished**; submitting a name whose task is still `pending`/`running` is rejected. Allowed characters: ASCII letters, digits, `.`, `_`, `-` (max 64). Omit for @@ -52,14 +52,15 @@ finishes within the window the full status is returned immediately. Useful to avoid a separate round-trip when the task is expected to finish soon. -Any `status` call (waited or not) that observes a terminal task clears -that task's completion todo — you already have the result in this -response, so no redundant loose-end follows. Narrow best-effort race: a -`status`/`run` inline wait that resolves in the same instant the task -actually finishes can still occasionally get both. +Any `status` call (waited or not) that observes a terminal task +suppresses that task's completion wake — you already have the result +in this response, so no redundant `bash-task-` inbox message +follows (#2270). Narrow best-effort race: a `status`/`run` inline wait +that resolves in the same instant the task actually finishes can still +occasionally get both. Tasks marked `interrupted` had their process killed by a harness -restart; a best-effort todo is still surfaced so the agent is not +restart; a best-effort wake was still sent so the agent is not silently blocked. Exposed as `mcp__bash__status`. @@ -67,8 +68,8 @@ Exposed as `mcp__bash__status`. ### `kill(id, force?)` Stop a running or pending task by its ID (from `run`). Fire-and-forget: -sends the signal and returns without waiting — the completion surfaces -in the agent's loose-ends; handle it on a future turn. +sends the signal and returns without waiting — handle the completion +wake (`from: "bash-task-"`) on a future turn. - `force: false` (default) — SIGINT to the task's **process group** (graceful; lets the process clean up). The whole process group is @@ -78,7 +79,7 @@ in the agent's loose-ends; handle it on a future turn. If a SIGINT'd task doesn't exit, call `kill` again with `force: true`. A still-pending task is cancelled before it starts. The task ends as -`killed` and surfaces in the loose-ends like any completion. +`killed` and fires the usual completion wake. Exposed as `mcp__bash__kill`. @@ -96,7 +97,7 @@ matrix MCP: - **`hive-bash-daemon`** — long-running process (one per agent container, systemd service in `nix/agent-modules/mcp.nix`). Owns subprocess management, - output file writing, and todo delivery on the harness's in-agent socket. + output file writing, `mcp-loose-ends/` state, and wake signal delivery. Listens on `/run/hive-bash/socket` inside the container. - **`hive-bash-mcp`** — stdio bridge spawned by claude per turn (declared @@ -107,17 +108,13 @@ matrix MCP: This split keeps claude's turn-local MCP bridge lightweight while the daemon tracks long-running tasks that outlive a single turn. -### Completion as a todo (loose-ends v2) +### Transient wake (bypass broker sqlite) -When a bash task changes state, `hive-bash-daemon` upserts a single keyed -todo (`key = task id`) on the harness's in-agent socket (`HIVE_AGENT_SOCKET`) -— "running" at start, then the completion summary when it finishes. The -summary change signals the harness turn loop directly (in-process, no broker -round-trip), so the agent is driven a turn to handle it via `get_loose_ends`, -then clears the todo with `mark_todo_done`. Same mechanism the matrix daemon -uses for unread rooms. An inline `wait_seconds` / `status` observation that -already delivered the result instead clears the keyed todo, so no redundant -loose-end follows. +When a bash task finishes, `hive-bash-daemon` sends the wake signal via +the agent's per-agent socket as a **transient wake** (`AgentRequest::Wake` +with `transient: true`). This bypasses broker sqlite for lower latency — +the same mechanism used by matrix events. The message is delivered +directly to the harness without touching the persistent message store. ## Relationship to the `execution` tool group diff --git a/docs/tools/matrix.md b/docs/tools/matrix.md index c81bed76..1c43bc02 100644 --- a/docs/tools/matrix.md +++ b/docs/tools/matrix.md @@ -90,16 +90,17 @@ returned by `get_loose_ends` so unread rooms surface in the loose-ends list between turns. **Invite wakes**: when the daemon's sync loop receives an -`m.room.member` invite event, it upserts a todo (keyed `invite:`) -on the harness's in-agent socket, which drives a turn. The daemon does -**not** auto-join — the agent calls `list_invites` to see pending +`m.room.member` invite event, it writes the invite to +`mcp-loose-ends/matrix.json` and fires a hyperhive wake. The daemon +does **not** auto-join — the agent calls `list_invites` to see pending invites and `resolve_invite` to accept or reject them. -**Pending invites as loose ends**: pending invites are upserted as -keyed todos and appear in `get_loose_ends` output as +**Pending invites as loose ends**: pending invites are written to +`mcp-loose-ends/matrix.json` and appear in `get_loose_ends` output as `[matrix] pending invite: — use list_invites to see, -resolve_invite to accept or reject`. The keyed todo is cleared when a -`resolve_invite` (or `join_room`) call resolves the invite. +resolve_invite to accept or reject`. The file is updated atomically +after each invite event and after each `resolve_invite` (or +`join_room`) call clears the entry. See [`docs/matrix.md`](../matrix.md) for the homeserver setup, provisioning flow, and federation config. diff --git a/hive-agent-mcp/src/loose_ends.rs b/hive-agent-mcp/src/loose_ends.rs new file mode 100644 index 00000000..7b6f4aad --- /dev/null +++ b/hive-agent-mcp/src/loose_ends.rs @@ -0,0 +1,42 @@ +//! 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; + +/// Resolution lives in `hive_sh4re::paths` so the harness + every MCP +/// daemon agree on where loose-end summary files are written. +fn loose_ends_dir() -> PathBuf { + hive_sh4re::paths::mcp_loose_ends_dir() +} + +/// 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-agent-mcp/src/main.rs b/hive-agent-mcp/src/main.rs index 2fbe4791..cc3c3c0d 100644 --- a/hive-agent-mcp/src/main.rs +++ b/hive-agent-mcp/src/main.rs @@ -18,6 +18,7 @@ use anyhow::Result; use clap::Parser; mod client; +mod loose_ends; mod mcp; mod paths; mod send_allow; diff --git a/hive-agent-mcp/src/mcp/mod.rs b/hive-agent-mcp/src/mcp/mod.rs index 73e30954..8acc2c6e 100644 --- a/hive-agent-mcp/src/mcp/mod.rs +++ b/hive-agent-mcp/src/mcp/mod.rs @@ -339,7 +339,20 @@ impl AgentServer { if is_self_query && let Some(todos) = local_todos().await { loose_ends.extend(todos); } - annotate_retries(render_loose_ends(&loose_ends), retries) + let mut out = annotate_retries(render_loose_ends(&loose_ends), retries); + // 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::loose_ends::collect(); + if !mcp_items.is_empty() { + use std::fmt::Write as _; + let n = mcp_items.len(); + let _ = write!(out, "\n\n{n} local task(s):"); + for item in &mcp_items { + let _ = write!(out, "\n- {item}"); + } + } + out }) .await } diff --git a/hive-agent/prompts/system.md b/hive-agent/prompts/system.md index 2ff4bca3..fc27896d 100644 --- a/hive-agent/prompts/system.md +++ b/hive-agent/prompts/system.md @@ -8,7 +8,7 @@ Tools (hyperhive surface): - (some agents only) **extra MCP tools** surfaced as `mcp____` — these are agent-specific (matrix client, scraper, db connector, etc.) declared in your `agent.nix` under `hyperhive.extraMcpServers`. Treat them as first-class tools alongside the hyperhive surface; the operator already auto-approved them at deploy time. - `mcp__hyperhive__ask(question, options?, multi?, ttl_seconds?, to?)` — surface a structured question to the human operator (default, or `to: "operator"`) OR a peer agent (`to: ""`). Returns immediately with a question id — do NOT wait inline. When the recipient answers, a system message with event `question_answered { id, question, answer, answerer }` lands in your inbox; handle it on a future turn. Use this for clarifications, permission for risky actions, choice between options, or peer Q&A without burning regular inbox slots. `options` is advisory: a short fixed-choice list when applicable, otherwise leave empty for free text. `multi: true` lets the answerer pick multiple (checkboxes), answer comes back comma-joined. `ttl_seconds` auto-cancels with answer `[expired]` (and `answerer: "ttl-watchdog"`) when the decision becomes moot. - `mcp__hyperhive__answer(id, answer)` — answer a question that was routed to YOU. You'll see one in your inbox as a `question_asked { id, asker, question, options, multi }` system event when a peer or the operator calls `ask(to: "", ...)`. The answer surfaces in the asker's inbox as a `question_answered` event. Strict authorisation: you can only answer questions where you are the declared target. -- `mcp__hyperhive__get_loose_ends(agent?)` — list your loose ends: unanswered questions where you're asker (waiting on someone) or target (owing a reply), reminders you've scheduled that haven't fired, plus in-container todos surfaced by your MCP daemons (finished/running background bash tasks, unread matrix rooms, forge threads). No args to list your own threads — cheap server-side sweep useful at turn start. Pass `agent: ""` to inspect a peer agent's threads. Direct child agents are always accessible. For non-children, the `query_agent_state` capability is required — without it the request is rejected with an error. +- `mcp__hyperhive__get_loose_ends(agent?)` — list your loose ends: unanswered questions where you're asker (waiting on someone) or target (owing a reply), plus reminders you've scheduled that haven't fired. No args to list your own threads — cheap server-side sweep useful at turn start. Pass `agent: ""` to inspect a peer agent's threads. Direct child agents are always accessible. For non-children, the `query_agent_state` capability is required — without it the request is rejected with an error. - `mcp__hyperhive__cancel_loose_end(kind, id)` — cancel one of your own open threads. `kind` is `"question"` (the asker — you, in this case — gets a `[cancelled by ]` answer so the waiter unblocks), `"reminder"` (hard-deleted before it fires), or `"approval"` (withdraws a pending approval you submitted that got superseded — root agent only; the server rejects this kind for all other callers). `id` from the matching `get_loose_ends` row or the original submission reply. - `mcp__hyperhive__remind(message, delay_seconds? | at_unix_timestamp?, file_path?)` — schedule a message to land in your _own_ inbox at a future time (sender shows as `reminder`). Set exactly one of `delay_seconds` (relative) or `at_unix_timestamp` (absolute). Use for self-paced follow-ups instead of blocking a whole turn on a long `recv` wait. A large `message` auto-spills to a file under `/agents/{label}/state/reminders/`; pass `file_path` to point at one yourself. Each agent's pending-reminder count is capped (default 50) — the tool will error if the cap is already reached. - `mcp__hyperhive__set_status(text)` — set a free-text status visible on the operator dashboard. **Call this at the start of every task** to say what you're working on (e.g. `"processing matrix messages"`, `"fixing #319 model priority"`, `"idle"`). Single line, ≤200 chars — the dashboard renders this as a short chip, so longer multi-line text is rejected. Pass an empty string to clear. Persists across harness restarts. diff --git a/hive-agent/src/forge_notify.rs b/hive-agent/src/forge_notify.rs index f45c6cd0..9f650877 100644 --- a/hive-agent/src/forge_notify.rs +++ b/hive-agent/src/forge_notify.rs @@ -64,8 +64,6 @@ 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, @@ -1066,40 +1064,33 @@ async fn poll_once( continue; }; - // Upsert a *todo* (loose-ends v2) on the harness's in-agent socket, - // keyed by the forge thread id, instead of firing a direct wake. A - // 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 req = hive_core_agent_sock::Request::Wake { + from: "forge".to_owned(), + body, }; - let deliver_result = crate::client::request::<_, hive_agent_sock::Response>(socket, &req) - .await - .map(|_| ()); + let deliver_result = + crate::client::request::<_, hive_core_agent_sock::Response>(socket, &req) + .await + .map(|_| ()); match deliver_result { Ok(()) => { - debug!(%id, "forge_notify: todo upserted"); - // Mark the thread read on forge immediately after upserting - // the todo. The todo is the durable work item now (it stays - // in `get_loose_ends` until the agent marks it done), so the - // forge unread flag no longer needs to track agent - // processing — clearing it keeps forge's unread set tiny by - // construction, so a container rebuild re-scan finds nothing - // stale (and idempotent re-upserts wouldn't re-wake anyway). - // The in-memory `delivered` entry below is only a - // within-process guard so a transient mark-read failure - // doesn't re-upsert next tick; it is deliberately NOT - // persisted — forge's own read-state is the cross-rebuild - // source of truth. + debug!(%id, "forge_notify: delivered"); + // Mark the thread read on forge immediately after a + // successful broker delivery. The broker inbox is the + // durable work queue now (each row has its own ack + // lifecycle), so the forge unread flag no longer needs to + // track agent processing — clearing it on delivery keeps + // forge's unread set tiny by construction, so a container + // rebuild re-scan finds nothing stale to re-deliver. The + // in-memory `delivered` entry below is only a within-process + // guard so a transient mark-read failure doesn't re-fire the + // wake next tick; it is deliberately NOT 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: todo upsert failed — leaving unread"); + warn!(%id, error = ?e, "forge_notify: deliver failed — leaving unread"); } } } diff --git a/hive-agent/src/main.rs b/hive-agent/src/main.rs index f167da3c..4c241696 100644 --- a/hive-agent/src/main.rs +++ b/hive-agent/src/main.rs @@ -421,15 +421,7 @@ async fn serve_main(socket: &Path, poll_ms: u64) -> Result<()> { for failure in plugins::install_configured().await { S::send_to_parent(socket, failure).await; } - // 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::from(hive_agent_sock::DEFAULT_AGENT_SOCKET), - std::path::PathBuf::from, - ), - )); + tokio::spawn(crate::forge_notify::run(socket.to_path_buf())); // 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. diff --git a/hive-bash-mcp/Cargo.toml b/hive-bash-mcp/Cargo.toml index 06269257..440e73a1 100644 --- a/hive-bash-mcp/Cargo.toml +++ b/hive-bash-mcp/Cargo.toml @@ -8,7 +8,7 @@ workspace = true [dependencies] anyhow.workspace = true -hive-agent-sock.workspace = true +hive-core-agent-sock.workspace = true hive-sh4re.workspace = true libc.workspace = true rmcp.workspace = true diff --git a/hive-bash-mcp/src/bin/mcp.rs b/hive-bash-mcp/src/bin/mcp.rs index 9c365a0a..69435cd9 100644 --- a/hive-bash-mcp/src/bin/mcp.rs +++ b/hive-bash-mcp/src/bin/mcp.rs @@ -125,7 +125,7 @@ fn render_bash_run(id: &str, resp: Result) -> String { /// immediately re-waiting on the same task. const BASH_IDLE_WAIT_HINT: &str = "\n\nThe task is still running — your wait timed out before it \ finished. If you have other useful work, do that and check back later (the task keeps running, and \ -it surfaces in your loose-ends when it completes) rather than immediately re-waiting."; +a wake fires when it completes) rather than immediately re-waiting."; /// Turn a `DaemonResponse` from a `BashStatus` call into the string /// claude sees as the tool result. When `waited` is set (the call @@ -154,7 +154,7 @@ fn render_bash_kill(resp: Result) -> String { let sig = payload["signal"].as_str().unwrap_or("SIGINT"); format!( "task `{id}`: {sig} sent to its process group; it transitions to `killed` \ - once the process exits and then surfaces in your loose-ends." + once the process exits and the usual completion wake fires." ) } else { format!("task `{id}` was pending — cancelled before it started.") @@ -180,16 +180,17 @@ struct BashRunArgs { timeout_secs: Option, /// Optional inline wait: `run` polls for up to `wait_seconds` /// (capped at 30) before returning. When the task finishes within the - /// window the full status is returned immediately (nothing to handle - /// later); when the timeout expires the task keeps running and the - /// normal `task started: id=` response is returned. Defaults to 3s. - /// Pass `0` to disable inline waiting and always get the immediate response. + /// window the full status is returned immediately and no wake is fired; + /// when the timeout expires the task keeps running and the normal + /// `task started: id=` response is returned. Defaults to 3s. Pass + /// `0` to disable inline waiting and always get the immediate response. #[serde(default = "default_wait")] wait_seconds: Option, /// Optional task name. When set it becomes the task id, so it appears - /// in `status` lookups and your loose-ends list — handy for recognising - /// a task later instead of an opaque hex id. A name can be reused once - /// its previous task has finished; reusing a name whose task is still + /// in the completion wake (`from: "bash-task-"`), in `status` + /// lookups, and in the loose-ends list — handy for recognising a task + /// later instead of an opaque hex id. A name can be reused once its + /// previous task has finished; reusing a name whose task is still /// running is rejected. Allowed chars: ASCII letters, digits, `.`, /// `_`, `-` (max 64). Omit to get the auto-generated id. #[serde(default)] @@ -236,18 +237,21 @@ struct BashMcp; impl BashMcp { #[tool( description = "Run a shell command in the background. Returns a task ID immediately — \ - do NOT wait inline. When the command finishes it surfaces as a todo in your \ - loose-ends (the exit status plus a `Read()` pointer to the captured \ - `.out`/`.err` files — read only what you need); handle it on a future turn. Use \ - `status` to poll within the same turn if needed. `timeout_secs` defaults to \ - `None` (no timeout) — task runs until natural exit; pass a value to kill after \ - N seconds. Pass `wait_seconds` (capped at 30) to wait inline for fast commands: \ - if the task finishes within the window you get the full status immediately \ - (nothing to handle later); otherwise it keeps running and you get \ - `task started: id=`. Defaults to 3; pass `0` to disable inline waiting. \ - Pass `name` to label the task with a memorable id (shown in `status` lookups \ - and your loose-ends) instead of an opaque hex id; reusable once the prior task \ - has finished, rejected while one is still running." + do NOT wait inline. When the command finishes, the harness fires a wake with \ + `from: \"bash-task-\"` carrying the exit status plus a `Read()` \ + pointer to the captured `.out`/`.err` files (not the output text itself — \ + read only what you need); handle it on a future turn. Use `status` to poll the task status within \ + the same turn if needed. `timeout_secs` defaults to `None` (no timeout) — \ + task runs until natural exit; pass an explicit value to kill after N seconds. \ + Pass `wait_seconds` (capped at 30) to wait inline for fast commands: when the \ + task finishes within the window the full status is returned immediately and no \ + wake is fired; when the timeout expires the task keeps running and the normal \ + `task started: id=` response is returned. `wait_seconds` defaults to 3; \ + pass `wait_seconds: 0` to disable inline waiting and always get the immediate \ + response. Pass `name` to label the task with a memorable id (used in the wake \ + `from`, `status` lookups, and the loose-ends list) instead of an opaque hex id; \ + a name is reusable once its prior task has finished, and rejected while one is \ + still running." )] async fn run(&self, Parameters(args): Parameters) -> String { let req = DaemonRequest::BashRun { @@ -293,8 +297,8 @@ impl BashMcp { signalled, so a runaway child (cargo/nix/etc.) is stopped too, not just the shell. \ Fire-and-forget: sends the signal and returns without waiting. If a SIGINT'd task \ doesn't exit, call kill again with `force: true` to SIGKILL. A still-pending task \ - is cancelled before it starts. The task ends as `killed` and surfaces in your \ - loose-ends like any completion." + is cancelled before it starts. The task ends as `killed` and fires the usual \ + completion wake." )] async fn kill(&self, Parameters(args): Parameters) -> String { let req = DaemonRequest::BashKill { diff --git a/hive-bash-mcp/src/main.rs b/hive-bash-mcp/src/main.rs index d7c1e166..8cfbe8ff 100644 --- a/hive-bash-mcp/src/main.rs +++ b/hive-bash-mcp/src/main.rs @@ -1,8 +1,7 @@ //! `hive-bash-daemon` binary — long-running per-agent bash task runner. //! Spawns `sh -c` subprocesses, monitors completion, writes task state -//! 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. +//! files, and fires hyperhive wake signals. Listens on a unix socket +//! for tool-call requests from the `hive-bash-mcp` stdio bridge. use anyhow::Result; @@ -16,17 +15,17 @@ async fn main() -> Result<()> { .init(); let socket_path = hive_bash_mcp::paths::daemon_socket(); - let todo_socket = hive_bash_mcp::paths::agent_socket(); + let wake_socket = hive_bash_mcp::paths::hyperhive_socket(); tracing::info!( socket = %socket_path.display(), - todo = %todo_socket.display(), + wake = %wake_socket.display(), "hive-bash-daemon starting" ); // Start the background runner loop — scans for pending tasks and - // spawns them, pushing todos to the harness on task transitions. - hive_bash_mcp::runner::spawn_loop(todo_socket); + // spawns them, sends wake signals on completion. + hive_bash_mcp::runner::spawn_loop(wake_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 478861fd..90714e4b 100644 --- a/hive-bash-mcp/src/paths.rs +++ b/hive-bash-mcp/src/paths.rs @@ -43,17 +43,21 @@ pub fn turn_stats_db() -> PathBuf { harness_dir().join("hyperhive-turn-stats.sqlite") } -/// 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. +/// 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`. #[must_use] -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, - ) +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() } /// 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 e17319fc..bba8dcb6 100644 --- a/hive-bash-mcp/src/runner.rs +++ b/hive-bash-mcp/src/runner.rs @@ -1,22 +1,15 @@ //! Bash subprocess runner: spawns `bash -c ` tasks, writes status -//! 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 `harness/bash-tasks/`, and fires hyperhive wake signals +//! on completion. //! //! 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): one keyed todo (`key = task id`) tracks the task -//! across its lifetime — upserted with a "running" summary at start, then -//! upserted again with the completion summary when it finishes. That summary -//! change signals the harness turn loop the same way the old completion wake -//! did; the agent clears the 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). The todo is still updated to -//! its interrupted-done summary so the agent is not silently blocked. +//! (the process died with the previous daemon). A best-effort wake is +//! still sent 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`. @@ -28,7 +21,6 @@ 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; @@ -165,82 +157,39 @@ pub fn read_task(id: &str) -> Option { } // --------------------------------------------------------------------------- -// Todo delivery (loose-ends v2) +// Loose-ends file (generic MCP loose-ends protocol) // --------------------------------------------------------------------------- -// -// 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: one keyed -// todo (`key = task id`) tracks the task across its whole lifetime. It's -// upserted with a stable "running" summary when the task starts (an -// idempotent no-op that never re-wakes), then upserted again with the -// completion summary when it finishes — that summary change signals the -// harness turn loop exactly like the old completion wake did. The agent -// reads the output and clears the todo with `mark_todo_done`. -/// 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"); - } +/// 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"); } -} - -/// Upsert the task's keyed todo (`key = id`) with `summary`. Used for both -/// the "running" surface at start and the "done" summary at completion — a -/// changed summary on the same row signals the turn loop, an unchanged -/// re-push is an idempotent no-op. -async fn upsert_bash_todo(socket: &Path, id: &str, summary: String) { - send_todo( - socket, - &TodoReq::UpsertTodo { - subsystem: "bash".to_owned(), - key: Some(id.to_owned()), - summary, - source: None, - }, - ) - .await; -} - -/// Clear a task's keyed todo — used when an inline `status`/`run` wait -/// already delivered the terminal result, so no `done` todo is warranted. -async fn clear_bash_todo(socket: &Path, id: &str) { - send_todo( - socket, - &TodoReq::ClearTodo { - subsystem: "bash".to_owned(), - key: Some(id.to_owned()), - all: false, - }, - ) - .await; } // --------------------------------------------------------------------------- @@ -316,10 +265,7 @@ pub fn submit_task(cmd: String, timeout_secs: Option, name: Option) stderr_tail: None, }; write_task(&task)?; - // 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). + refresh_loose_ends(); Ok(id) } @@ -439,9 +385,7 @@ pub fn kill_task(id: &str, force: bool) -> (bool, bool) { task.status = TaskStatus::Killed; task.completed_at = Some(now_unix()); let _ = write_task(&task); - // 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. + refresh_loose_ends(); return (true, false); } (false, false) @@ -451,9 +395,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 harness's in-agent todo socket (`HIVE_AGENT_SOCKET`), where task -/// transitions are pushed as todos. Call once at daemon startup. +/// 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. pub fn spawn_loop(socket: PathBuf) { tokio::spawn(async move { run_loop(socket).await; @@ -499,12 +443,8 @@ 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"); } - upsert_bash_todo( - socket, - &id, - done_summary(&id, "interrupted (daemon restarted)", None), - ) - .await; + refresh_loose_ends(); + send_wake(socket, &id, "interrupted (daemon restarted)", None).await; } } @@ -554,12 +494,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"); } - upsert_bash_todo( - socket, - &id, - format!("bash task `{id}` running: `{}`", task.cmd), - ) - .await; + refresh_loose_ends(); // 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. @@ -627,33 +562,25 @@ 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"); } - // If an inline `status`/`run` wait already handed this exact terminal - // result to the caller in a tool response (see `wait_for_task` / - // `observe_terminal`), there's nothing left to surface — retire the - // keyed todo without a `done` upsert so the agent gets no redundant - // wake. Narrow race: an inline waiter polling in the few hundred ms - // around this point may lose the race and still get a `done` todo - // alongside its inline result; best-effort, same tolerance as the rest - // of this daemon's guarantees. + refresh_loose_ends(); + + // Skip the wake 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. if take_wake_suppressed(&id) { - clear_bash_todo(socket, &id).await; - tracing::debug!(id = %id, "bash_runner: done todo suppressed (already observed via status)"); + tracing::debug!(id = %id, "bash_runner: wake suppressed (already observed via status)"); return; } - // Transition the SAME keyed todo from "running" to its done summary: the - // summary change signals the turn loop, and the agent clears the todo - // with `mark_todo_done` after reading the output. Pass only whether each - // stream produced (trimmed) output — the summary carries a `Read()` - // pointer to the captured `.out`/`.err` files rather than inlining the tail. + // 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. let has_stdout = !stdout_tail.as_deref().unwrap_or("").trim().is_empty(); let has_stderr = !stderr_tail.as_deref().unwrap_or("").trim().is_empty(); - upsert_bash_todo( - socket, - &id, - done_summary(&id, &summary, Some((has_stdout, has_stderr))), - ) - .await; + send_wake(socket, &id, &summary, Some((has_stdout, has_stderr))).await; } /// Run `bash -c cmd` in its own process group, streaming output to files. @@ -799,20 +726,24 @@ fn tail_file(path: &Path, max_bytes: usize) -> Option { } // --------------------------------------------------------------------------- -// Completion summary +// Wake delivery // --------------------------------------------------------------------------- -/// 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 { +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)>, +) { 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) @@ -829,7 +760,38 @@ fn done_summary(id: &str, summary: &str, output: Option<(bool, bool)>) -> String ); } } - body + 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"); + } + } } #[cfg(test)] diff --git a/hive-sh4re/src/paths.rs b/hive-sh4re/src/paths.rs index 1a390d73..9a3609e7 100644 --- a/hive-sh4re/src/paths.rs +++ b/hive-sh4re/src/paths.rs @@ -28,3 +28,10 @@ pub fn harness_dir() -> PathBuf { let label = std::env::var("HIVE_LABEL").unwrap_or_default(); PathBuf::from(format!("/agents/{label}/harness")) } + +/// Directory where out-of-process MCP daemons write loose-end summary +/// files (`.json`) for the harness to scan in `get_loose_ends`. +#[must_use] +pub fn mcp_loose_ends_dir() -> PathBuf { + harness_dir().join("mcp-loose-ends") +} diff --git a/nix/agent-modules/mcp.nix b/nix/agent-modules/mcp.nix index 07fab953..6a6aaeca 100644 --- a/nix/agent-modules/mcp.nix +++ b/nix/agent-modules/mcp.nix @@ -175,11 +175,7 @@ in ]; environment = { HIVE_BASH_SOCKET = "/run/hive-bash/socket"; - # 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"; + HIVE_CONTROL_SOCKET = "/run/hive/mcp.sock"; RUST_LOG = "info"; # HYPERHIVE_HARNESS_DIR and HYPERHIVE_STATE_DIR are already # injected via systemd.globalEnvironment by the meta flake