From 9bcd6976fe3486e007f9e7ee8a266c84f79eb2d4 Mon Sep 17 00:00:00 2001 From: damocles Date: Sun, 31 May 2026 14:24:07 +0200 Subject: [PATCH] harness: converge agent + manager turn paths (#692) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit three shared helpers replace the duplicated pre-#598 patterns: - `log_system_event` lifts the HelperEvent parse + bus emit out of handle_manager_turn so agents log QuestionAnswered/ContainerCrash/ reparent notifications the same way (#692 part 1). - `format_turn_failure` produces the failure-notification body using identity::qualified_label() instead of a label param threaded through three layers. drops `label` from handle_agent_turn, agent_serve_loop, agent_check_and_inject_continue. - `consume_continue_sentinel` lifts the file-probe so both surfaces reuse it (#692 part 2 — sentinel now works for manager too). agent_notify_manager_of_failure → agent_notify_parent_of_failure: routes via the sentinel landed in #703 instead of the literal string 'manager'. mirrored on manager side; root-manager failures resolve to operator via topology::resolve_recipient. handle_*_turn signatures now identical modulo the wire-type prefix (part 3 acceptance from the issue). --- hive-ag3nt/src/bin/hive.rs | 150 ++++++++++++++++++++++++++++++------- 1 file changed, 121 insertions(+), 29 deletions(-) diff --git a/hive-ag3nt/src/bin/hive.rs b/hive-ag3nt/src/bin/hive.rs index d518e1e1..cb9749dc 100644 --- a/hive-ag3nt/src/bin/hive.rs +++ b/hive-ag3nt/src/bin/hive.rs @@ -109,6 +109,55 @@ async fn main() -> Result<()> { } } +// ---------- shared turn helpers (#692) ---------- + +/// Surface a `SYSTEM_SENDER` message in the live event bus + tracing +/// log. Was manager-only pre-#692; agents receive `QuestionAnswered`, +/// `ContainerCrash`, reparent notifications, and friends the same way +/// the manager does, so the parse + log path is identical too. Quiet +/// no-op when `from` isn't `SYSTEM_SENDER`. +fn log_system_event(bus: &Bus, from: &str, body: &str) { + if from != SYSTEM_SENDER { + return; + } + let parsed = serde_json::from_str::(body).ok(); + if let Some(event) = parsed { + tracing::info!(?event, "helper event"); + } else { + tracing::info!(%from, %body, "system message"); + } + bus.emit(LiveEvent::Note { text: format!("[system] {body}") }); +} + +/// Body string for the turn-failure notification we route to +/// `` on `TurnOutcome::Failed`. Reads the hive-qualified +/// identity so the receiver sees `agent@hive` rather than relying on +/// the caller threading a `label` through every turn-handling layer. +/// Falls back to `` when `HIVE_LABEL` is missing so a +/// misconfigured harness still produces a parseable line. +fn format_turn_failure(err: &anyhow::Error) -> String { + let who = hive_ag3nt::identity::qualified_label(); + let who = if who.is_empty() { "".to_owned() } else { who }; + 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 = hive_ag3nt::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 +} + // ---------- agent role ---------- async fn agent_serve_main(socket: &Path, poll_ms: u64) -> Result<()> { @@ -156,7 +205,6 @@ async fn agent_serve_main(socket: &Path, poll_ms: u64) -> Result<()> { stats, &files, turn_lock, - &label, ) .await } @@ -171,7 +219,6 @@ async fn agent_serve_main(socket: &Path, poll_ms: u64) -> Result<()> { stats, &files, turn_lock, - &label, ) .await } @@ -188,7 +235,6 @@ async fn agent_serve_loop( stats: Option, files: &turn::TurnFiles, turn_lock: TurnLock, - label: &str, ) -> Result<()> { tracing::info!(socket = %socket.display(), "hive-ag3nt serve"); agent_requeue_inflight(socket).await; @@ -205,7 +251,7 @@ async fn agent_serve_loop( Ok(AgentResponse::Messages { messages }) if !messages.is_empty() => { let first = messages.into_iter().next().expect("checked non-empty"); let auth_failed = - handle_agent_turn(socket, &bus, stats.as_ref(), files, &turn_lock, label, first) + handle_agent_turn(socket, &bus, stats.as_ref(), files, &turn_lock, first) .await; if auth_failed { *login_state.lock().unwrap() = LoginState::NeedsLogin; @@ -249,12 +295,12 @@ async fn handle_agent_turn( stats: Option<&TurnStats>, files: &turn::TurnFiles, turn_lock: &TurnLock, - label: &str, first: hive_sh4re::DeliveredMessage, ) -> bool { let from = first.from; let body = first.body; let redelivered = first.redelivered; + log_system_event(bus, &from, &body); tracing::info!(%from, %body, %redelivered, "inbox"); let unread = agent_inbox_unread(socket).await; bus.emit(LiveEvent::TurnStart { from: from.clone(), body: body.clone(), unread }); @@ -292,7 +338,7 @@ async fn handle_agent_turn( agent_requeue_inflight(socket).await; } if let turn::TurnOutcome::Failed(e) = &outcome { - agent_notify_manager_of_failure(socket, label, e).await; + agent_notify_parent_of_failure(socket, e).await; } if let Some(stats) = stats { let ended_at = serve_common::now_unix(); @@ -316,7 +362,7 @@ async fn handle_agent_turn( if pending > 0 { tracing::info!(%pending, "pending messages after turn; fetching next"); } - agent_check_and_inject_continue(socket, label).await; + agent_check_and_inject_continue(socket).await; matches!(outcome, turn::TurnOutcome::AuthFailed) } @@ -346,19 +392,24 @@ async fn agent_requeue_inflight(socket: &Path) { } } -async fn agent_notify_manager_of_failure(socket: &Path, label: &str, err: &anyhow::Error) { - let body = format!("[system] agent `{label}` claude turn failed:\n{err:#}"); +/// Notify whoever's structurally watching this agent that the claude +/// turn failed. Pre-#692 this was `agent_notify_manager_of_failure` +/// targeting the literal string `"manager"`; now it routes through +/// the `` sentinel landed in #703 so failures bubble to the +/// real parent (and to operator for root agents). Body identity +/// comes from `format_turn_failure` — no `label` plumbing. +async fn agent_notify_parent_of_failure(socket: &Path, err: &anyhow::Error) { let res = client::request::<_, AgentResponse>( socket, &AgentRequest::Send { - to: "manager".into(), - body, + to: hive_sh4re::PARENT_RECIPIENT.into(), + body: format_turn_failure(err), in_reply_to: None, }, ) .await; if let Err(e) = res { - tracing::warn!(error = ?e, "failed to notify manager of turn failure"); + tracing::warn!(error = ?e, "failed to notify parent of turn failure"); } } @@ -387,13 +438,8 @@ async fn agent_post_turn_counts(socket: &Path) -> (Option, Option) { (threads, reminders) } -async fn agent_check_and_inject_continue(socket: &Path, label: &str) { - let sentinel = hive_ag3nt::paths::state_dir().join("hyperhive-continue"); - if !sentinel.exists() { - return; - } - if let Err(e) = std::fs::remove_file(&sentinel) { - tracing::warn!(error = %e, "check_and_inject_continue: remove sentinel failed"); +async fn agent_check_and_inject_continue(socket: &Path) { + if !consume_continue_sentinel() { return; } let res = client::request::<_, AgentResponse>( @@ -406,7 +452,7 @@ async fn agent_check_and_inject_continue(socket: &Path, label: &str) { .await; match res { Ok(AgentResponse::Ok) => { - tracing::info!(%label, "request_next_turn: injected self-continue wake"); + tracing::info!("request_next_turn: injected self-continue wake"); } Ok(AgentResponse::Err { message }) => { tracing::warn!(%message, "check_and_inject_continue: wake rejected"); @@ -579,15 +625,7 @@ async fn handle_manager_turn( let from = first.from; let body = first.body; let redelivered = first.redelivered; - if from == SYSTEM_SENDER { - let parsed = serde_json::from_str::(&body).ok(); - if let Some(event) = parsed { - tracing::info!(?event, "helper event"); - } else { - tracing::info!(%from, %body, "system message"); - } - bus.emit(LiveEvent::Note { text: format!("[system] {body}") }); - } + log_system_event(bus, &from, &body); tracing::info!(%from, %body, %redelivered, "manager inbox"); let unread = manager_inbox_unread(socket).await; bus.emit(LiveEvent::TurnStart { from: from.clone(), body: body.clone(), unread }); @@ -624,6 +662,9 @@ async fn handle_manager_turn( tracing::warn!("auth-failed; parking until re-login"); manager_requeue_inflight(socket).await; } + if let turn::TurnOutcome::Failed(e) = &outcome { + manager_notify_parent_of_failure(socket, e).await; + } if let Some(stats) = stats { let ended_at = serve_common::now_unix(); let duration_ms = @@ -646,9 +687,60 @@ async fn handle_manager_turn( if pending > 0 { tracing::info!(%pending, "pending messages after turn; fetching next"); } + manager_check_and_inject_continue(socket).await; matches!(outcome, turn::TurnOutcome::AuthFailed) } +/// Manager mirror of `agent_notify_parent_of_failure`. For a root +/// manager (`topology::parent_of("manager")` is `None`) the +/// `` sentinel resolves to `operator`, which surfaces the +/// failure in the dashboard T4LK box — the only audience above +/// the manager that can act on it. +async fn manager_notify_parent_of_failure(socket: &Path, err: &anyhow::Error) { + let res = client::request::<_, ManagerResponse>( + socket, + &ManagerRequest::Send { + to: hive_sh4re::PARENT_RECIPIENT.into(), + body: format_turn_failure(err), + in_reply_to: None, + }, + ) + .await; + if let Err(e) = res { + tracing::warn!(error = ?e, "failed to notify parent of turn failure"); + } +} + +/// Manager mirror of `agent_check_and_inject_continue`. The +/// `request_next_turn` MCP tool drops the same sentinel from either +/// flavor; both harnesses pick it up the same way and fire their +/// role-specific `Wake` request. +async fn manager_check_and_inject_continue(socket: &Path) { + if !consume_continue_sentinel() { + return; + } + let res = client::request::<_, ManagerResponse>( + socket, + &ManagerRequest::Wake { + from: "self".into(), + body: "continue".into(), + }, + ) + .await; + match res { + Ok(ManagerResponse::Ok) => { + tracing::info!("request_next_turn: injected self-continue wake"); + } + Ok(ManagerResponse::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 manager_ack_turn(socket: &Path) { match client::request::<_, ManagerResponse>(socket, &ManagerRequest::AckTurn).await { Ok(ManagerResponse::Ok) => {}