diff --git a/docs/tools/bash.md b/docs/tools/bash.md index 72d03ff0..c28dc5b0 100644 --- a/docs/tools/bash.md +++ b/docs/tools/bash.md @@ -16,7 +16,8 @@ Stdout and stderr stream to `harness/bash-tasks/.{out,err}`. 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. +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). @@ -51,6 +52,13 @@ 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 +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 wake was still sent so the agent is not silently blocked. diff --git a/hive-bash-mcp/src/runner.rs b/hive-bash-mcp/src/runner.rs index ac1915b1..e1718c04 100644 --- a/hive-bash-mcp/src/runner.rs +++ b/hive-bash-mcp/src/runner.rs @@ -63,6 +63,39 @@ fn running() -> &'static Mutex> { RUNNING.get_or_init(|| Mutex::new(HashMap::new())) } +// --------------------------------------------------------------------------- +// Wake suppression: skip the completion wake when a caller already +// synchronously observed the task's terminal state via an inline +// `wait_seconds` poll on `BashRun` or `BashStatus` — the tool response +// already delivered the result in that same turn, so a follow-up wake +// message would just be a redundant duplicate of information the agent has. +// --------------------------------------------------------------------------- + +/// In-memory only — a daemon restart wipes it, which is fine: a task still +/// `running` across a restart is marked `interrupted` on boot (see module +/// docs) and gets its own fresh wake, independent of this set. +fn wake_suppressed() -> &'static Mutex> { + static SUPPRESSED: OnceLock>> = OnceLock::new(); + SUPPRESSED.get_or_init(|| Mutex::new(HashSet::new())) +} + +/// Mark `id`'s completion wake as already-delivered-inline. Called after an +/// inline `wait_seconds` poll (on `BashRun` or `BashStatus`) observes a +/// terminal task, before the response carrying the full status is written +/// back to the caller. Idempotent — safe to call more than once per id. +pub(crate) fn suppress_wake(id: &str) { + wake_suppressed().lock().unwrap().insert(id.to_owned()); +} + +/// Consume (remove + report) `id`'s suppression flag. Returns `true` if the +/// wake should be skipped. One-shot: a task id is only ever completed once, +/// so there's no risk of a stale suppression leaking onto a later task with +/// the same id (names are only reusable once the prior task has finished, +/// i.e. after this has already been consumed). +fn take_wake_suppressed(id: &str) -> bool { + wake_suppressed().lock().unwrap().remove(id) +} + /// Outcome of one `exec_cmd` run. enum ExecOutcome { /// Process exited on its own with this status code. @@ -259,10 +292,16 @@ pub fn active_tasks() -> Vec { /// Inline wait: poll `read_task(id)` until terminal state or deadline. /// Returns the final task on success, or `None` if it never completed. +/// +/// Whenever a terminal task is observed here, the caller is about to receive +/// that result directly in its tool response — so the completion wake for +/// `id` is marked [`suppress_wake`]d: a status query (waited or not) that +/// already told the agent the outcome shouldn't be followed by a redundant +/// "task finished" inbox message for the same information. pub async fn wait_for_task(id: &str, wait_secs: u64) -> Option { let cap = wait_secs.min(MAX_WAIT_SECS); if cap == 0 { - return read_task(id); + return observe_terminal(read_task(id)); } let deadline = tokio::time::Instant::now() + Duration::from_secs(cap); loop { @@ -276,6 +315,7 @@ pub async fn wait_for_task(id: &str, wait_secs: u64) -> Option { | TaskStatus::Interrupted | TaskStatus::Killed ) { + suppress_wake(id); return Some(task); } } @@ -285,7 +325,22 @@ pub async fn wait_for_task(id: &str, wait_secs: u64) -> Option { } tokio::time::sleep(Duration::from_millis(POLL_MS)).await; } - read_task(id) + observe_terminal(read_task(id)) +} + +/// Marks the wake suppressed if `task` is in a terminal state; passes +/// `task` through unchanged either way. Shared tail helper for both +/// `wait_for_task` return points. +fn observe_terminal(task: Option) -> Option { + if let Some(t) = &task + && matches!( + t.status, + TaskStatus::Done | TaskStatus::TimedOut | TaskStatus::Interrupted | TaskStatus::Killed + ) + { + suppress_wake(&t.id); + } + task } /// Kill a running or still-pending task. @@ -507,6 +562,17 @@ async fn run_task(mut task: TaskFile, socket: &Path) { } 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) { + tracing::debug!(id = %id, "bash_runner: wake suppressed (already observed via status)"); + return; + } + let out_snippet = stdout_tail.as_deref().unwrap_or("").trim(); let err_snippet = stderr_tail.as_deref().unwrap_or("").trim(); send_wake(socket, &id, &summary, Some((out_snippet, err_snippet))).await; @@ -746,4 +812,31 @@ mod tests { // Exactly at the cap is allowed. assert!(validate_task_name(&"x".repeat(MAX_TASK_NAME_LEN)).is_ok()); } + + // Wake suppression is a single process-wide registry (see + // `wake_suppressed()`), so these run serially against distinct ids to + // avoid cross-test interference under parallel test execution. + + #[test] + fn wake_suppression_is_one_shot() { + use super::{suppress_wake, take_wake_suppressed}; + let id = "test-2270-one-shot"; + assert!(!take_wake_suppressed(id), "unset id starts unsuppressed"); + suppress_wake(id); + assert!(take_wake_suppressed(id), "set id reports suppressed once"); + assert!( + !take_wake_suppressed(id), + "consuming the flag clears it — second read is unsuppressed" + ); + } + + #[test] + fn wake_suppression_is_idempotent_to_set() { + use super::{suppress_wake, take_wake_suppressed}; + let id = "test-2270-idempotent"; + suppress_wake(id); + suppress_wake(id); // simulates two concurrent observers of the same terminal task + assert!(take_wake_suppressed(id)); + assert!(!take_wake_suppressed(id)); + } }