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

fix(#1543): drive request_next_turn continue in-process, not via the broker

Browse source
This commit is contained in:
damocles 2026-06-09 00:12:14 +02:00 • committed by mara
commit daa6a59324
2 changed files with 144 additions and 59 deletions
Download patch file Download diff file Expand all files Collapse all files

18
docs/turn-loop.md
View file

@ -82,7 +82,7 @@ zero-sized impl (`AgentSurface`) wrapping:
- One async method per wire op: `ack_turn`, `requeue_inflight`,
`inbox_unread`, `post_turn_counts`, `send_to_parent`,
`self_wake`, `recv_next`, `wake_external`.
`recv_next`, `wake_external`.
`main()` calls `serve_main::<AgentSurface>` for all roles. The turn
loop (`serve_loop` / `handle_turn` / `wake`) has no per-role branches.
@ -117,9 +117,19 @@ and the manager fall through to operator).
After the outcome handler, the stats sink records a row and the
`hyperhive-continue` sentinel (dropped by the `request_next_turn`
MCP tool) is consumed if present, firing `self_wake` so the next
turn starts with `{ from: "self", body: "continue" }` even if the
inbox is empty.
MCP tool) is consumed if present. `handle_turn` reports the result
to `serve_loop` via `TurnControl { auth_failed, continue_requested,
pending }`. When a continue was requested, the turn did not
auth-fail, and the inbox is empty (`pending == 0`), `serve_loop`
drives the next turn in-process with a synthetic
`{ from: "self", body: "continue" }` message (`synthetic_continue`)
— it never goes through the broker, so the self-continue doesn't
persist to sqlite or show up as a recv'able inbox message. If real
messages are already pending the continue is dropped: those messages
drive the next turn(s) via `recv_next`, so an explicit self-wake
isn't needed (this is the `request_next_turn` contract — "no effect
if a new inbox message arrives before this turn ends"). The
`should_self_continue` predicate encodes exactly that decision.
## The claude invocation

185
hive-ag3nt/src/bin/hive.rs
View file

@ -130,6 +130,49 @@ fn consume_continue_sentinel() -> bool {
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 (#1543).
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
/// (#1543). `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,
}
}
// ---------- surface trait ----------
/// What a `Recv` long-poll returned. Decoupled from the per-role
@ -173,10 +216,6 @@ trait Surface {
/// agents/manager fall through to operator).
fn send_to_parent(socket: &Path, body: String) -> impl Future<Output = ()>;
/// Fire a `Wake { from: "self", body: "continue" }` at our own
/// inbox — the `request_next_turn` sentinel pickup.
fn self_wake(socket: &Path) -> impl Future<Output = ()>;
/// Long-poll the broker for the next message. Wraps the
/// `Messages`/empty/error trichotomy in `RecvOutcome` so the
/// generic `serve_loop` doesn't need the per-role Response enum
@ -264,30 +303,6 @@ impl Surface for AgentSurface {
}
}
async fn self_wake(socket: &Path) {
let res = client::request::<_, AgentResponse>(
socket,
&AgentRequest::Wake {
from: "self".into(),
body: "continue".into(),
transient: false,
},
)
.await;
match res {
Ok(AgentResponse::Ok) => {
tracing::info!("request_next_turn: injected self-continue wake");
}
Ok(AgentResponse::Err { message }) => {
tracing::warn!(%message, "check_and_inject_continue: wake rejected");
}
Err(e) => {
tracing::warn!(error = ?e, "check_and_inject_continue: wake transport error");
}
_ => {}
}
}
async fn recv_next(socket: &Path) -> RecvOutcome {
let recv: Result<AgentResponse> = client::request(
socket,
@ -439,29 +454,40 @@ async fn serve_loop<S: Surface>(
) -> Result<()> {
tracing::info!(socket = %socket.display(), "harness serve");
S::requeue_inflight(socket).await;
// 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 (#1543). Never
// persisted: it lives entirely in this loop's stack.
let mut self_continue: Option<hive_sh4re::DeliveredMessage> = None;
loop {
match S::recv_next(socket).await {
RecvOutcome::Message(first) => {
let auth_failed =
handle_turn::<S>(socket, &bus, stats.as_ref(), files, &turn_lock, first).await;
if auth_failed {
*login_state.lock().unwrap() = LoginState::NeedsLogin;
turn::wait_for_login(
&claude_dir,
login_state.clone(),
&bus,
u64::try_from(interval.as_millis()).unwrap_or(2000),
)
.await;
let next = match self_continue.take() {
Some(msg) => msg,
None => match S::recv_next(socket).await {
RecvOutcome::Message(first) => first,
RecvOutcome::Empty => {
tokio::time::sleep(interval).await;
continue;
}
}
RecvOutcome::Empty => {
tokio::time::sleep(interval).await;
}
RecvOutcome::TransportError => {
// `recv_next` already logged the detail; just retry.
// No backoff: the long-poll wait is itself the throttle.
}
RecvOutcome::TransportError => {
// `recv_next` already logged the detail; just retry.
// No backoff: the long-poll wait is itself the throttle.
continue;
}
},
};
let ctrl = handle_turn::<S>(socket, &bus, stats.as_ref(), files, &turn_lock, next).await;
if ctrl.auth_failed {
*login_state.lock().unwrap() = LoginState::NeedsLogin;
turn::wait_for_login(
&claude_dir,
login_state.clone(),
&bus,
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());
}
}
}
@ -469,8 +495,9 @@ async fn serve_loop<S: Surface>(
/// 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 true iff the outcome was
/// `AuthFailed` — the caller flips the harness to needs-login.
/// 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.
async fn handle_turn<S: Surface>(
socket: &Path,
bus: &Bus,
@ -478,7 +505,7 @@ async fn handle_turn<S: Surface>(
files: &turn::TurnFiles,
turn_lock: &TurnLock,
first: hive_sh4re::DeliveredMessage,
) -> bool {
) -> TurnControl {
let from = first.from;
let body = first.body;
let redelivered = first.redelivered;
@ -550,10 +577,11 @@ async fn handle_turn<S: Surface>(
if pending > 0 {
tracing::info!(%pending, "pending messages after turn; fetching next");
}
if consume_continue_sentinel() {
S::self_wake(socket).await;
TurnControl {
auth_failed: matches!(outcome, turn::TurnOutcome::AuthFailed),
continue_requested: consume_continue_sentinel(),
pending,
}
matches!(outcome, turn::TurnOutcome::AuthFailed)
}
/// External `hive wake` subcommand — push a message into our own
@ -569,3 +597,50 @@ async fn wake<S: Surface>(socket: &Path, from: String, body: String) -> Result<(
};
S::wake_external(socket, from, body).await
}
#[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 (#1543 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());
}
}