mara/hyperhive
mara/hyperhive
Watch
0
0
Fork
You've already forked hyperhive
 0
Code Activity

hive-bash-mcp: send queued output ahead of a status

Browse source
This commit is contained in:
damocles 2026-11-15 17:57:47 +01:00
commit a4b2ef0814
2 changed files with 11 additions and 7 deletions
Download patch file Download diff file Expand all files Collapse all files

3
docs/4-implementation/agent-tools/bash.md
View file

@ -99,7 +99,8 @@ log line, and a task never waits on the queue.
`$SWARM.term.<agent>.bash.<task>`, each with the `Swarm-Stream` header `$SWARM.term.<agent>.bash.<task>`, each with the `Swarm-Stream` header
(`stdout` or `stderr`). A chunk never splits a character, and one larger than (`stdout` or `stderr`). A chunk never splits a character, and one larger than
the queue's payload limit goes out in pieces. When the process exits, the daemon the queue's payload limit goes out in pieces. When the process exits, the daemon
publishes what the files still hold before it writes the final status. queues what the files still hold before it writes the final status, and the
publish task sends queued output ahead of a status.
## Relationship to the `execution` permission ## Relationship to the `execution` permission

15
hive-bash-mcp/src/swarm_term.rs
View file

@ -203,13 +203,10 @@ async fn run(
let period = Duration::from_secs(bash_task::HEARTBEAT_S.unsigned_abs()); let period = Duration::from_secs(bash_task::HEARTBEAT_S.unsigned_abs());
let mut beat = tokio::time::interval_at(tokio::time::Instant::now() + period, period); let mut beat = tokio::time::interval_at(tokio::time::Instant::now() + period, period);
loop { loop {
// Output first: a task's last chunks are queued before its final status,
// and taking queued output ahead of a status keeps that order on the wire.
tokio::select! { tokio::select! {
Some(event) = status_rx.recv() => { biased;
if let Event::Status { task, status } = event {
apply(&mut live, &task, &status);
publish_status(&mut queue, &target, &task, &status).await;
}
}
event = output_rx.recv() => { event = output_rx.recv() => {
let Some(Event::Output { task, stream, text }) = event else { return }; let Some(Event::Output { task, stream, text }) = event else { return };
let missed = dropped.swap(0, Ordering::Relaxed); let missed = dropped.swap(0, Ordering::Relaxed);
@ -218,6 +215,12 @@ async fn run(
} }
publish_output(&mut queue, &target, &task, stream, &text).await; publish_output(&mut queue, &target, &task, stream, &text).await;
} }
Some(event) = status_rx.recv() => {
if let Event::Status { task, status } = event {
apply(&mut live, &task, &status);
publish_status(&mut queue, &target, &task, &status).await;
}
}
_ = beat.tick() => { _ = beat.tick() => {
for (task, status) in refresh(&mut live, unix_now()) { for (task, status) in refresh(&mut live, unix_now()) {
publish_status(&mut queue, &target, &task, &status).await; publish_status(&mut queue, &target, &task, &status).await;