vinzenz/hyperhive
vinzenz/hyperhive
Watch
0
0
Fork
You've already forked hyperhive
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

refactor(#2628): one keyed todo per task transitions running→done (mara: same id changes to done, not a keyless one-off)

Browse source
This commit is contained in:
damocles 2026-07-22 00:15:39 +02:00
commit 4c36343f04
1 changed files with 48 additions and 53 deletions
Download patch file Download diff file Expand all files Collapse all files

101
hive-bash-mcp/src/runner.rs
View file

@ -7,15 +7,16 @@
//! - `<id>.out` — captured stdout (streamed while running)
//! - `<id>.err` — captured stderr (streamed while running)
//!
//! Todos (loose-ends v2): a keyed todo (`key = task id`) is upserted while
//! a task is active and cleared when it finishes; a keyless one-off `done`
//! todo then carries the completion summary, which signals the harness turn
//! loop the same way the old completion wake did. The agent clears a `done`
//! todo with `mark_todo_done` once it has read the output.
//! 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). A best-effort `done` todo
//! is still pushed so the agent is not silently blocked.
//! (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 runner kills the child process on timeout — `tokio::process::Child::drop()`
//! does not kill children, so we explicitly call `child.kill().await`.
@ -168,11 +169,13 @@ pub fn read_task(id: &str) -> Option<TaskFile> {
// ---------------------------------------------------------------------------
//
// 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: a keyed
// todo (`key = task id`) is upserted while the task is active and cleared
// when it finishes, and a keyless one-off `done` todo carries the
// completion summary. A fresh keyless row always signals the harness turn
// loop, so it drives a turn exactly like the old wake did.
// 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
@ -209,24 +212,26 @@ async fn send_todo(socket: &Path, req: &TodoReq) {
}
}
/// Upsert the keyed "active" todo for a running task (surfaces it in
/// `get_loose_ends`). The summary is stable across the task's lifetime, so
/// re-pushing it is an idempotent no-op that never re-wakes.
async fn upsert_active_todo(socket: &Path, id: &str, cmd: &str) {
/// 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: format!("bash task `{id}` running: `{cmd}`"),
summary,
source: None,
},
)
.await;
}
/// Clear a task's keyed "active" todo once it has reached a terminal state.
async fn clear_active_todo(socket: &Path, id: &str) {
/// 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 {
@ -238,23 +243,6 @@ async fn clear_active_todo(socket: &Path, id: &str) {
.await;
}
/// Push a keyless one-off `done` todo carrying the completion `summary` —
/// the loose-ends-v2 replacement for the old completion wake. A fresh
/// keyless row always signals the turn loop, so the agent is driven a turn
/// to read the output (and clears the todo with `mark_todo_done`).
async fn push_done_todo(socket: &Path, summary: String) {
send_todo(
socket,
&TodoReq::UpsertTodo {
subsystem: "bash".to_owned(),
key: None,
summary,
source: None,
},
)
.await;
}
// ---------------------------------------------------------------------------
// Public API used by daemon dispatch
// ---------------------------------------------------------------------------
@ -511,9 +499,9 @@ 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");
}
clear_active_todo(socket, &id).await;
push_done_todo(
upsert_bash_todo(
socket,
&id,
done_summary(&id, "interrupted (daemon restarted)", None),
)
.await;
@ -566,7 +554,12 @@ 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_active_todo(socket, &id, &task.cmd).await;
upsert_bash_todo(
socket,
&id,
format!("bash task `{id}` running: `{}`", task.cmd),
)
.await;
// 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.
@ -634,28 +627,30 @@ 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");
}
// Always retire the keyed "active" todo — the task has finished
// regardless of whether its completion also gets a `done` todo below.
clear_active_todo(socket, &id).await;
// Skip the `done` todo 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 `done` todo alongside its inline result;
// best-effort, same tolerance as the rest of this daemon's guarantees.
// 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.
if take_wake_suppressed(&id) {
clear_bash_todo(socket, &id).await;
tracing::debug!(id = %id, "bash_runner: done todo suppressed (already observed via status)");
return;
}
// Pass only whether each stream produced (trimmed) output — the `done`
// todo carries a `Read(<path>)` pointer to the captured `.out`/`.err`
// files rather than inlining the tail into the summary.
// 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(<path>)`
// pointer to the captured `.out`/`.err` files rather than inlining the tail.
let has_stdout = !stdout_tail.as_deref().unwrap_or("").trim().is_empty();
let has_stderr = !stderr_tail.as_deref().unwrap_or("").trim().is_empty();
push_done_todo(
upsert_bash_todo(
socket,
&id,
done_summary(&id, &summary, Some((has_stdout, has_stderr))),
)
.await;