From 2b5edef4d14b36ccf34aeac6f0f561aab3dfe131 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?m=C3=BCde?= Date: Sun, 5 Jul 2026 21:42:13 +0200 Subject: [PATCH] fix(agent): recover PromptTooLong via archive+requeue; model TurnOutcome as Result --- hive-ag3nt/src/bin/hive.rs | 23 ++++++---- hive-ag3nt/src/serve_common.rs | 14 +++--- hive-ag3nt/src/turn.rs | 82 +++++++++++++++++++++++----------- 3 files changed, 76 insertions(+), 43 deletions(-) diff --git a/hive-ag3nt/src/bin/hive.rs b/hive-ag3nt/src/bin/hive.rs index 37cf3864..9fbd5b30 100644 --- a/hive-ag3nt/src/bin/hive.rs +++ b/hive-ag3nt/src/bin/hive.rs @@ -108,7 +108,7 @@ fn log_system_event(bus: &Bus, from: &str, body: &str) { } /// Body string for the turn-failure notification we route to -/// `` on `TurnOutcome::Failed`. Reads the hive-qualified +/// `` on `TurnError::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 @@ -601,13 +601,10 @@ async fn handle_turn( let outcome = turn::drive_turn(&prompt, files, bus, session).await; turn::emit_turn_end(bus, &outcome); bus.set_state(TurnState::Idle); - if matches!( - outcome, - turn::TurnOutcome::Ok | turn::TurnOutcome::Compacted - ) { + if outcome.is_ok() { S::ack_turn(socket).await; } - if matches!(outcome, turn::TurnOutcome::RateLimited) { + if matches!(outcome, Err(turn::TurnError::RateLimited)) { let secs = turn::rate_limit_sleep_secs(); bus.emit_status("rate_limited"); bus.emit(LiveEvent::Note { @@ -618,7 +615,7 @@ async fn handle_turn( S::requeue_inflight(socket).await; bus.emit_status("online"); } - if matches!(outcome, turn::TurnOutcome::AuthFailed) { + if matches!(outcome, Err(turn::TurnError::AuthFailed)) { bus.emit_status("needs_login_idle"); bus.emit(LiveEvent::Note { text: "API 401 — waiting for re-login via web UI".into(), @@ -626,7 +623,15 @@ async fn handle_turn( tracing::warn!("auth-failed; parking until re-login"); S::requeue_inflight(socket).await; } - if let turn::TurnOutcome::Failed(e) = &outcome { + if matches!(outcome, Err(turn::TurnError::PromptTooLong)) { + // `drive_turn` already archived the session; requeue the message so it + // redelivers into the fresh session (which fits — the wake prompt is + // tiny, the overflow was the now-cleared context). No status park: the + // agent is healthy, it just needs one more delivery. + tracing::warn!("prompt-too-long; session archived, requeueing message for a fresh turn"); + S::requeue_inflight(socket).await; + } + if let Err(turn::TurnError::Failed(e)) = &outcome { S::send_to_parent(socket, format_turn_failure(e)).await; } if let Some(stats) = stats { @@ -659,7 +664,7 @@ async fn handle_turn( tracing::info!(%pending, "pending messages after turn; fetching next"); } TurnControl { - auth_failed: matches!(outcome, turn::TurnOutcome::AuthFailed), + auth_failed: matches!(outcome, Err(turn::TurnError::AuthFailed)), continue_requested: consume_continue_sentinel(), pending, } diff --git a/hive-ag3nt/src/serve_common.rs b/hive-ag3nt/src/serve_common.rs index 3a3b00cc..e23bd4c0 100644 --- a/hive-ag3nt/src/serve_common.rs +++ b/hive-ag3nt/src/serve_common.rs @@ -5,7 +5,7 @@ use crate::events::Bus; use crate::mcp::REDELIVERY_HINT; -use crate::turn::TurnOutcome; +use crate::turn::{TurnError, TurnOutcome}; use crate::turn_stats::TurnStatRow; /// Assemble the per-turn wake prompt string. The role/tools/etc. live in the @@ -100,12 +100,12 @@ pub fn build_row(args: TurnRowArgs<'_>) -> TurnStatRow { serde_json::to_string(&tool_calls).ok() }; let (result_kind, note) = match outcome { - TurnOutcome::Ok => ("ok", None), - TurnOutcome::Compacted => ("compacted", None), - TurnOutcome::PromptTooLong => ("prompt_too_long", None), - TurnOutcome::RateLimited => ("rate_limited", None), - TurnOutcome::AuthFailed => ("auth_failed", None), - TurnOutcome::Failed(e) => ("failed", Some(format!("{e:#}"))), + Ok(false) => ("ok", None), + Ok(true) => ("compacted", None), + Err(TurnError::PromptTooLong) => ("prompt_too_long", None), + Err(TurnError::RateLimited) => ("rate_limited", None), + Err(TurnError::AuthFailed) => ("auth_failed", None), + Err(TurnError::Failed(e)) => ("failed", Some(format!("{e:#}"))), }; let wake_from = if wake_from.starts_with("bash-task-") { "bash-task".to_owned() diff --git a/hive-ag3nt/src/turn.rs b/hive-ag3nt/src/turn.rs index 62661fbe..9d9384cd 100644 --- a/hive-ag3nt/src/turn.rs +++ b/hive-ag3nt/src/turn.rs @@ -139,29 +139,39 @@ pub async fn write_system_prompt(socket: &Path, label: &str) -> Result crate::prompt::write_system_prompt(socket, label).await } -/// One claude turn's outcome. The harness uses this to decide whether to -/// transparently kick off a compaction and retry. +/// One claude turn's outcome: `Ok(compacted)` on success, or a [`TurnError`] +/// the serve loop must act on. The `compacted` bool is `true` when a +/// compaction ran this turn (reactively on overflow, or proactively per the +/// policy — or an operator `/compact` at turn end); it's recorded as +/// `result_kind = "compacted"` in turn stats so the stats page can distinguish +/// those turns. Both `Ok(true)` and `Ok(false)` are ack'd; the error cases +/// each map to a distinct serve-loop action (see [`emit_turn_end`] and the +/// `hive` serve loop). +pub type TurnOutcome = std::result::Result; + +/// The ways a turn can end without a usable result. Each is deliberately *not* +/// a generic failure — the serve loop reacts to each differently (requeue, +/// park, escalate). #[derive(Debug)] -pub enum TurnOutcome { - Ok, - /// Turn completed and proactive context-size compaction fired afterwards. - /// Treated like `Ok` for ack and failure-notification purposes; recorded - /// as `result_kind = "compacted"` in turn stats so the stats page can - /// distinguish normal turns from turns that triggered a compaction. - Compacted, +pub enum TurnError { /// claude saw "Prompt is too long" and even a reactive compact + retry /// (inside [`InfiniteSession::run`]) couldn't bring it back under the - /// window. Rare; the serve loop treats it like `Ok` (acks the turn). + /// window. Rare. [`drive_turn`] archives the session (so the next turn + /// starts fresh) and the serve loop requeues the in-flight message, which + /// redelivers into that fresh session — the wake prompt itself is tiny, so + /// the overflow was the accumulated context, which the archive clears. PromptTooLong, /// The Anthropic API refused the request due to a rate limit, per-account /// usage cap, or exhausted credit balance. The serve loop should park for - /// `rate_limit_sleep_secs()` and retry — NOT bubble up as a crash. + /// `rate_limit_sleep_secs()` and requeue — NOT bubble up as a crash. RateLimited, /// The Anthropic API rejected the request with 401 (OAuth session /// expired or revoked). The serve loop should flip the container /// into `needs_login_idle` and stop driving turns until the /// operator re-auths via the per-agent web UI. AuthFailed, + /// A hard failure with no recovery — the serve loop escalates it to the + /// parent (`send_to_parent`). Failed(anyhow::Error), } @@ -309,24 +319,33 @@ pub async fn drive_turn( text: format!("created fresh session titled \"{}\"", session_title()), }); } - if progress.compacted { - TurnOutcome::Compacted - } else { - TurnOutcome::Ok - } + Ok(progress.compacted) } Err(e) => error_to_turn(e), }; + if matches!(outcome, Err(TurnError::PromptTooLong)) { + // The lib already compacted + retried and the session is still over the + // window. Archive it here (session lifecycle stays hive-side) so the + // requeued message — handled by the serve loop — redelivers into a + // fresh session that fits. + bus.emit(LiveEvent::Note { + text: "context still over the window after compaction — archiving session so the \ + retried message starts fresh" + .into(), + }); + archive_session(bus); + return Err(TurnError::PromptTooLong); + } // Operator `/compact` (`POST /api/compact`) deferred to the turn boundary: // run it now that the turn is done, so it works mid-turn rather than only // when the agent is idle. Only on a healthy turn — no point spawning a // compaction after a rate-limited / auth-failed / crashed one. - if bus.take_compact() && matches!(outcome, TurnOutcome::Ok | TurnOutcome::Compacted) { + if bus.take_compact() && outcome.is_ok() { bus.emit(LiveEvent::Note { text: "operator: /compact — running at turn end".into(), }); let _ = session.compact(&config, &sink).await; - return TurnOutcome::Compacted; + return Ok(true); } outcome } @@ -374,28 +393,35 @@ fn maybe_auto_reset(bus: &Bus) { /// semantics stay consistent across every agent role. pub fn emit_turn_end(bus: &Bus, outcome: &TurnOutcome) { match outcome { - TurnOutcome::Ok | TurnOutcome::Compacted | TurnOutcome::PromptTooLong => { + Ok(_) => { bus.emit(LiveEvent::TurnEnd { ok: true, note: None, }); tracing::info!("turn finished"); } - TurnOutcome::RateLimited => { + Err(TurnError::PromptTooLong) => { + bus.emit(LiveEvent::TurnEnd { + ok: false, + note: Some("context too long after compaction — session archived, retrying".into()), + }); + tracing::warn!("turn prompt-too-long; archived session and requeueing"); + } + Err(TurnError::RateLimited) => { bus.emit(LiveEvent::TurnEnd { ok: false, note: Some("rate limited — parking until quota resets".into()), }); tracing::warn!("turn rate-limited"); } - TurnOutcome::AuthFailed => { + Err(TurnError::AuthFailed) => { bus.emit(LiveEvent::TurnEnd { ok: false, note: Some("authentication failed (401) — waiting for re-login".into()), }); tracing::warn!("turn auth-failed (401)"); } - TurnOutcome::Failed(e) => { + Err(TurnError::Failed(e)) => { let note = format!("{e:#}"); bus.emit(LiveEvent::TurnEnd { ok: false, @@ -503,11 +529,13 @@ fn claude_config(bus: &Bus, files: &TurnFiles) -> Config { fn error_to_turn(err: hive_claude::Error) -> TurnOutcome { use hive_claude::Error; match err { - Error::PromptTooLong => TurnOutcome::PromptTooLong, - Error::RateLimited => TurnOutcome::RateLimited, - Error::AuthFailed => TurnOutcome::AuthFailed, - Error::SessionNotFound => TurnOutcome::Ok, - other => TurnOutcome::Failed(other.into()), + Error::PromptTooLong => Err(TurnError::PromptTooLong), + Error::RateLimited => Err(TurnError::RateLimited), + Error::AuthFailed => Err(TurnError::AuthFailed), + // A resume-miss the lib couldn't self-heal is benign — treat it as a + // clean (non-compacted) turn; the next turn creates the session fresh. + Error::SessionNotFound => Ok(false), + other => Err(TurnError::Failed(other.into())), } }