diff --git a/docs/tools/bash.md b/docs/tools/bash.md index c28dc5b0..72d03ff0 100644 --- a/docs/tools/bash.md +++ b/docs/tools/bash.md @@ -16,8 +16,7 @@ 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 — unless `wait_seconds` already delivered the terminal result -inline, in which case the wake is suppressed (see `status` below). +turn. * `timeout_secs` — kill the task after N seconds and mark it `timed_out`. Omit for no timeout (runs until natural exit). @@ -52,13 +51,6 @@ 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 d7228012..ac1915b1 100644 --- a/hive-bash-mcp/src/runner.rs +++ b/hive-bash-mcp/src/runner.rs @@ -63,41 +63,6 @@ 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 by intent: a completed task's entry is -/// meant to be drained exactly once. There's a known low-probability race -/// (see `run_task`'s completion handler) where `suppress_wake` for a task -/// fires *after* this has already run for it — if the same name gets reused -/// before that late insert lands, the dangling flag could suppress the new -/// task's wake instead. Not eliminated, just narrow. -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. @@ -294,16 +259,10 @@ 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 observe_terminal(read_task(id)); + return read_task(id); } let deadline = tokio::time::Instant::now() + Duration::from_secs(cap); loop { @@ -317,7 +276,6 @@ pub async fn wait_for_task(id: &str, wait_secs: u64) -> Option { | TaskStatus::Interrupted | TaskStatus::Killed ) { - suppress_wake(id); return Some(task); } } @@ -327,22 +285,7 @@ pub async fn wait_for_task(id: &str, wait_secs: u64) -> Option { } tokio::time::sleep(Duration::from_millis(POLL_MS)).await; } - 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 + read_task(id) } /// Kill a running or still-pending task. @@ -564,17 +507,6 @@ 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; @@ -814,31 +746,4 @@ 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)); - } }