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

harness: converge agent + manager turn paths (#692)

Browse source 
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 <parent> 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).
This commit is contained in:
damocles 2026-05-31 14:24:07 +02:00 • committed by mara
commit 9bcd6976fe
1 changed files with 121 additions and 29 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -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::<HelperEvent>(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
/// `<parent>` 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 `<unknown>` 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() { "<unknown>".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<TurnStats>,
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 `<parent>` 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<u64>, Option<u64>) {
(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::<HelperEvent>(&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
/// `<parent>` 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) => {}