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

remove request_next_turn: same-turn continuation is always worse than an external wake

Browse source
This commit is contained in:
damocles 2026-07-27 22:07:40 +02:00 • committed by mara
commit fffe0a2c29
8 changed files with 103 additions and 253 deletions
Download patch file Download diff file Expand all files Collapse all files

6
hive-agent/prompts/system.md
View file

@ -2,7 +2,7 @@ You are hyperhive agent `{label}` (qualified: `{qualified_label}`){hive_identity
Tools (hyperhive surface). Full signature + behavior for each comes from the tool's own MCP description (you already received it via the MCP tool schema) — this is just the map of what exists and which ones are gated, so you know where to look:
- **Inbox / messaging** (always available): `mcp__hyperhive__recv`, `mcp__hyperhive__ack_until`, `mcp__hyperhive__send`, `mcp__hyperhive__ask`, `mcp__hyperhive__answer`, `mcp__hyperhive__get_loose_ends`, `mcp__hyperhive__cancel_loose_end`, `mcp__hyperhive__remind`, `mcp__hyperhive__set_status`, `mcp__hyperhive__get_agent_meta`, `mcp__hyperhive__request_next_turn`. Two habits worth internalizing beyond the tool descriptions themselves: prefer ending the turn over parking in `recv` when idle (only turn-boundaries observe in-container todo wakes — bash-task completions, matrix unread, forge activity — and ending the turn is also your checkpoint); and `ask`/`answer` are async — `ask` returns immediately with a question id, the reply lands later as a `question_answered` system event, never block a turn waiting on it inline.
- **Inbox / messaging** (always available): `mcp__hyperhive__recv`, `mcp__hyperhive__ack_until`, `mcp__hyperhive__send`, `mcp__hyperhive__ask`, `mcp__hyperhive__answer`, `mcp__hyperhive__get_loose_ends`, `mcp__hyperhive__cancel_loose_end`, `mcp__hyperhive__remind`, `mcp__hyperhive__set_status`, `mcp__hyperhive__get_agent_meta`. Two habits worth internalizing beyond the tool descriptions themselves: prefer ending the turn over parking in `recv` when idle (only turn-boundaries observe in-container todo wakes — bash-task completions, matrix unread, forge activity — and ending the turn is also your checkpoint); and `ask`/`answer` are async — `ask` returns immediately with a question id, the reply lands later as a `question_answered` system event, never block a turn waiting on it inline.
- **Extra MCP tools** (some agents only): `mcp__<server>__<tool>` — agent-specific (matrix client, scraper, db connector, etc.) declared in your `agent.nix` under `hyperhive.extraMcpServers`. First-class tools, already operator-approved at deploy time.
- **Lifecycle** (_requires `lifecycle` tool group_, direct children only, no approval needed): `restart`, `kill`, `start`, `update`, `list_containers`.
- **Approvals** (_requires `approvals` tool group_, queues an operator approval): `request_init_config`, `request_apply_commit`, `request_update_meta_inputs`.
@ -42,6 +42,6 @@ Keep messages short — a few sentences each. For anything big (file listings, l
When your inbox has a message, handle it and stop. Don't narrate intent — act.
**Turns are your checkpoint.** The harness runs one claude turn per inbox message; when you stop, it acknowledges that message and your `--continue` session is saved to disk. Ending the turn is how you commit progress — both the session and the inbox acknowledgement. If the container restarts _while a turn is still running_, the message that drove it was never acknowledged, so it gets redelivered on the next boot, prefixed `[redelivered after harness restart — may already be handled]`. A long single turn that does step after step widens the window where a restart loses work and forces that redelivery, so prefer short turns: do a unit of work, write anything durable under `/agents/{label}/state/`, and end.
**Turns are your checkpoint.** The harness runs one claude turn per inbox message; when you stop, it acknowledges that message and your `--continue` session is saved to disk. Ending the turn is how you commit progress — both the session and the inbox acknowledgement. If the container restarts _while a turn is still running_, the message that drove it was never acknowledged, so it gets redelivered on the next boot, prefixed `[redelivered after harness restart — may already be handled]`. A long single turn that does step after step widens the window where a restart loses work and forces that redelivery, so prefer short turns: do a unit of work, write anything durable under `/agents/{label}/state/` **at every natural boundary, not just when a turn happens to end**, and stop.
**To keep working without waiting for a new message, call `request_next_turn()`** before you stop. The harness immediately starts a fresh turn with `from: "self"`, `body: "continue"` — the supported way to run multi-step work (long builds, sequential edits) as a series of checkpointed turns rather than one monolithic turn. Don't busy-wait inside a turn for a condition to resolve: end the turn and let the next wake drive the continuation — a `remind` you scheduled, an external event, a backgrounded bash task's completion, or `request_next_turn()`. (A long-poll `recv(wait_seconds: …)` blocks _within_ the current turn — it parks for new inbox messages but does not end the turn or checkpoint, so it isn't a substitute for ending the turn.)
For multi-step work (long builds, sequential edits) that spans more than one turn: end the turn and let an external wake drive the next one — a new inbox message, a `remind` you scheduled, or a backgrounded bash task's completion. Ending the turn is never a same-turn continuation: it's the checkpoint itself, and the only place an in-container todo wake (bash-task completion, matrix unread, forge activity) can reach you — a long-poll `recv(wait_seconds: …)` blocks _within_ the current turn instead and misses exactly those wakes, so it isn't a substitute for ending the turn. **Keep `set_status` current whenever the work changes** — a stale status is what actually loses context across a restart, not turn length; a status set at the start of a task and never touched again is a bug, not a shortcut.

264
hive-agent/src/main.rs
View file

@ -188,72 +188,19 @@ fn format_turn_failure(err: &anyhow::Error) -> String {
format!("[system] `{who}` claude turn failed:\n{err:#}")
}
/// Check for the `hyperhive-continue` sentinel under the state dir
/// (dropped by the `request_next_turn` MCP tool). Returns true and
/// consumes the file when present; false otherwise. Caller fires
/// the role-specific `Wake` request — the sentinel itself is wire-
/// agnostic so this helper lives outside both surfaces.
fn consume_continue_sentinel() -> bool {
let sentinel = crate::paths::state_dir().join("hyperhive-continue");
if !sentinel.exists() {
return false;
}
if let Err(e) = std::fs::remove_file(&sentinel) {
tracing::warn!(error = %e, "consume_continue_sentinel: remove sentinel failed");
return false;
}
true
}
/// What a finished turn tells the serve loop to do next. Replaces the
/// bare `auth_failed` bool so the loop can also act on a pending
/// `request_next_turn` without round-tripping a synthetic message
/// through the broker.
/// What a finished turn tells the serve loop to do next.
struct TurnControl {
/// The turn ended in `AuthFailed` — caller parks on login.
auth_failed: bool,
/// `request_next_turn` was called during the turn (the
/// `hyperhive-continue` sentinel was dropped + consumed).
continue_requested: bool,
/// Inbox unread count observed right after the turn. Used to
/// decide whether a self-continue is actually needed.
pending: u64,
}
/// Decide whether the serve loop should drive a self-continue turn
/// in-process. A continue is only "needed" when nothing else will
/// wake the agent: if real messages are already pending they drive
/// the next turn(s) and the continue is dropped (matches the
/// `request_next_turn` contract — "no effect if a new inbox message
/// arrives before this turn ends"). Auth-failed parks the loop on
/// login, so it suppresses the continue too.
fn should_self_continue(ctrl: &TurnControl) -> bool {
ctrl.continue_requested && !ctrl.auth_failed && ctrl.pending == 0
}
/// Synthesize the `from: "self"` / `body: "continue"` message that a
/// `request_next_turn` self-continue drives. Built in-process rather
/// than fetched from the broker — it never touches the send/recv
/// path, so it doesn't persist to sqlite or pollute the inbox.
/// `id = 0` is a non-broker sentinel: the synthetic message
/// has no DB row, and `AckTurn` keys off the recipient's in-flight
/// list (which is empty here) rather than this id.
fn synthetic_continue() -> hive_sh4re::DeliveredMessage {
hive_sh4re::DeliveredMessage {
from: "self".into(),
body: "continue".into(),
id: 0,
redelivered: false,
in_reply_to: None,
}
}
/// Synthesize the message that drives a turn when an in-container producer
/// upserted a new/changed *todo* over the in-agent socket (loose-ends v2).
/// The harness owns the todo store locally and signals the serve loop
/// directly — so this wake never touches the broker (no long-poll, no
/// marker file). `id = 0` is the same non-broker sentinel as
/// [`synthetic_continue`].
/// marker file). `id = 0` is a non-broker sentinel: the synthetic message
/// has no DB row, and `AckTurn` keys off the recipient's in-flight list
/// (which is empty here) rather than this id.
fn synthetic_todo_message() -> hive_sh4re::DeliveredMessage {
hive_sh4re::DeliveredMessage {
from: "todo".into(),
@ -678,20 +625,12 @@ async fn serve_loop<S: Surface>(
// The durable claude session, built once and reused for every turn +
// idle compaction below (it's effectively stateless).
let session = turn::make_session(&bus);
// Set when a turn calls `request_next_turn` and no real work is
// pending — the next iteration drives this synthetic message
// in-process instead of long-polling the broker. Never
// persisted: it lives entirely in this loop's stack.
let mut self_continue: Option<hive_sh4re::DeliveredMessage> = None;
// Tracks the last observed pause state so the transitions get logged
// once each instead of twelve lines a minute while parked.
let mut was_paused = false;
loop {
// Pause gate. While the marker is present this loop drives no
// turns at all — deliberately *before* the `self_continue.take()`
// below, so a `request_next_turn` that raced the pause is still
// waiting when the agent resumes rather than being consumed by
// a turn that never runs.
// turns at all.
//
// Nothing here touches the broker: not calling `S::recv_next` is
// exactly the "messages queue unacked, resume drains the
@ -722,78 +661,75 @@ async fn serve_loop<S: Surface>(
});
was_paused = false;
}
let next = match self_continue.take() {
Some(msg) => msg,
None => match {
// Idle wait: race the broker long-poll against a local
// todo signal so an in-container producer's upsert drives a
// turn without any broker round-trip. `biased` polls the
// broker recv first, so a genuinely-ready inbox message is
// never dropped in favour of the todo wake.
tokio::select! {
biased;
o = S::recv_next(socket) => o,
() = todo_wake.notified() => RecvOutcome::LocalTodo,
Some(dm) = reminder_rx.recv() => RecvOutcome::Message(dm),
}
} {
RecvOutcome::Message(first) => first,
RecvOutcome::LocalTodo => {
// Gate on `has_any()` before spawning a turn: a burst of
// same-turn upserts can arm a second `Notify` permit that
// outlives the turn which already drained its payload
// (the phantom-todo-wake issue) — `notify_one` doesn't
// coalesce once the first permit's been consumed, so the surplus wake
// fires the instant the loop is back here even though
// there's nothing left to show. Fail open (drive a turn
// anyway) on a `has_any` error so a flaky sqlite read
// never silently swallows a real wake.
let has_any = todos_store
.as_ref()
.is_none_or(|store| store.has_any().unwrap_or(true));
if !has_any {
tracing::debug!("todo wake fired against an empty store — stale, skipping");
continue;
}
tracing::debug!("todo wake consumed, sending synthetic todo message");
synthetic_todo_message()
}
RecvOutcome::Empty => {
// Idle: no message this poll. Service a queued operator
// `/compact` here so it runs even when no turn is driving
// (the in-flight case is handled at the end of drive_turn).
let compacted = turn::run_pending_compact(files, &bus, &session).await;
if !compacted {
tokio::time::sleep(interval).await;
}
let next = match {
// Idle wait: race the broker long-poll against a local
// todo signal so an in-container producer's upsert drives a
// turn without any broker round-trip. `biased` polls the
// broker recv first, so a genuinely-ready inbox message is
// never dropped in favour of the todo wake.
tokio::select! {
biased;
o = S::recv_next(socket) => o,
() = todo_wake.notified() => RecvOutcome::LocalTodo,
Some(dm) = reminder_rx.recv() => RecvOutcome::Message(dm),
}
} {
RecvOutcome::Message(first) => first,
RecvOutcome::LocalTodo => {
// Gate on `has_any()` before spawning a turn: a burst of
// same-turn upserts can arm a second `Notify` permit that
// outlives the turn which already drained its payload
// (the phantom-todo-wake issue) — `notify_one` doesn't
// coalesce once the first permit's been consumed, so the surplus wake
// fires the instant the loop is back here even though
// there's nothing left to show. Fail open (drive a turn
// anyway) on a `has_any` error so a flaky sqlite read
// never silently swallows a real wake.
let has_any = todos_store
.as_ref()
.is_none_or(|store| store.has_any().unwrap_or(true));
if !has_any {
tracing::debug!("todo wake fired against an empty store — stale, skipping");
continue;
}
RecvOutcome::TransportError => {
// `recv_next` already logged the detail; just retry.
// No backoff: the long-poll wait is itself the throttle.
continue;
tracing::debug!("todo wake consumed, sending synthetic todo message");
synthetic_todo_message()
}
RecvOutcome::Empty => {
// Idle: no message this poll. Service a queued operator
// `/compact` here so it runs even when no turn is driving
// (the in-flight case is handled at the end of drive_turn).
let compacted = turn::run_pending_compact(files, &bus, &session).await;
if !compacted {
tokio::time::sleep(interval).await;
}
RecvOutcome::GracefulStop => {
// c0re fenced our inbox and wants a clean stop. Run one
// checkpoint turn so the agent flushes durable /state,
// report completion, then exit the loop → the harness
// process ends and the container can be stopped.
tracing::info!(
"graceful stop signalled — running stop-checkpoint turn, then exiting"
);
let _ = handle_turn::<S>(
socket,
&bus,
stats.as_ref(),
files,
&session,
graceful_stop_message(),
)
.await;
S::graceful_stop_complete(socket).await;
return Ok(());
}
},
continue;
}
RecvOutcome::TransportError => {
// `recv_next` already logged the detail; just retry.
// No backoff: the long-poll wait is itself the throttle.
continue;
}
RecvOutcome::GracefulStop => {
// c0re fenced our inbox and wants a clean stop. Run one
// checkpoint turn so the agent flushes durable /state,
// report completion, then exit the loop → the harness
// process ends and the container can be stopped.
tracing::info!(
"graceful stop signalled — running stop-checkpoint turn, then exiting"
);
let _ = handle_turn::<S>(
socket,
&bus,
stats.as_ref(),
files,
&session,
graceful_stop_message(),
)
.await;
S::graceful_stop_complete(socket).await;
return Ok(());
}
};
let ctrl = handle_turn::<S>(socket, &bus, stats.as_ref(), files, &session, next).await;
if ctrl.auth_failed {
@ -805,19 +741,14 @@ async fn serve_loop<S: Surface>(
u64::try_from(interval.as_millis()).unwrap_or(2000),
)
.await;
} else if should_self_continue(&ctrl) {
tracing::info!("request_next_turn: driving self-continue turn in-process");
self_continue = Some(synthetic_continue());
}
}
}
/// Drive a single turn: emit boot-of-turn events, run claude, ack on
/// success / requeue on rate-limit-or-401 / notify parent on failure,
/// record stats, then pick up the `request_next_turn` sentinel if it's
/// been dropped during the turn. Returns a `TurnControl` carrying the
/// auth-failed flag, whether a self-continue was requested, and the
/// post-turn inbox count — the serve loop decides what to do next.
/// record stats. Returns a `TurnControl` carrying the auth-failed flag —
/// the serve loop decides what to do next.
async fn handle_turn<S: Surface>(
socket: &Path,
bus: &Bus,
@ -933,54 +864,5 @@ async fn handle_turn<S: Surface>(
}
TurnControl {
auth_failed: matches!(outcome, Err(turn::TurnError::AuthFailed)),
continue_requested: consume_continue_sentinel(),
pending,
}
}
#[cfg(test)]
mod continue_tests {
use super::{TurnControl, should_self_continue, synthetic_continue};
fn ctrl(auth_failed: bool, continue_requested: bool, pending: u64) -> TurnControl {
TurnControl {
auth_failed,
continue_requested,
pending,
}
}
#[test]
fn self_continue_when_requested_and_inbox_empty() {
assert!(should_self_continue(&ctrl(false, true, 0)));
}
#[test]
fn no_self_continue_when_not_requested() {
assert!(!should_self_continue(&ctrl(false, false, 0)));
}
#[test]
fn no_self_continue_when_real_messages_pending() {
// A real message will drive the next turn via recv — the
// continue is superseded, not needed (request_next_turn contract).
assert!(!should_self_continue(&ctrl(false, true, 3)));
}
#[test]
fn no_self_continue_when_auth_failed() {
// Auth-failed parks the loop on login; a queued continue must
// not jump the gate.
assert!(!should_self_continue(&ctrl(true, true, 0)));
}
#[test]
fn synthetic_continue_shape() {
let m = synthetic_continue();
assert_eq!(m.from, "self");
assert_eq!(m.body, "continue");
assert_eq!(m.id, 0);
assert!(!m.redelivered);
assert!(m.in_reply_to.is_none());
}
}

1
hive-agent/src/stream_enrich.rs
View file

@ -362,7 +362,6 @@ fn tool_icon(name: &str) -> &'static str {
"mcp__hyperhive__cancel_loose_end" => "✂️",
"mcp__hyperhive__ack_until" => "✅",
"mcp__hyperhive__get_agent_meta" => "ℹ️",
"mcp__hyperhive__request_next_turn" => "⏩",
"mcp__hyperhive__restart" => "↻",
"mcp__hyperhive__kill" => "⏹️",
"mcp__hyperhive__start" => "▶️",