diff --git a/docs/turn-loop.md b/docs/turn-loop.md index 42524014..f8ece2d3 100644 --- a/docs/turn-loop.md +++ b/docs/turn-loop.md @@ -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::` 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 diff --git a/hive-ag3nt/src/bin/hive.rs b/hive-ag3nt/src/bin/hive.rs index 5193631f..ca58567e 100644 --- a/hive-ag3nt/src/bin/hive.rs +++ b/hive-ag3nt/src/bin/hive.rs @@ -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; - /// Fire a `Wake { from: "self", body: "continue" }` at our own - /// inbox — the `request_next_turn` sentinel pickup. - fn self_wake(socket: &Path) -> impl Future; - /// 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 = client::request( socket, @@ -439,29 +454,40 @@ async fn serve_loop( ) -> 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 = None; loop { - match S::recv_next(socket).await { - RecvOutcome::Message(first) => { - let auth_failed = - handle_turn::(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::(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( /// 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( socket: &Path, bus: &Bus, @@ -478,7 +505,7 @@ async fn handle_turn( 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( 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(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()); + } +}