From b4f54a194b25ca4e804fe1619859e93611c6251f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?m=C3=BCde?= Date: Mon, 6 Jul 2026 00:01:15 +0200 Subject: [PATCH] fix(agent): don't eat /compact flag on failed turn; restore ctx fallback; requeue on SessionNotFound --- docs/turn-loop.md | 1 + hive-ag3nt/src/bin/hive.rs | 7 +++++++ hive-ag3nt/src/serve_common.rs | 1 + hive-ag3nt/src/turn.rs | 37 ++++++++++++++++++++++++++++------ 4 files changed, 40 insertions(+), 6 deletions(-) diff --git a/docs/turn-loop.md b/docs/turn-loop.md index 0b70e486..30b5390f 100644 --- a/docs/turn-loop.md +++ b/docs/turn-loop.md @@ -113,6 +113,7 @@ else a `TurnError`) drives the post-claude branch: | `Err(PromptTooLong)` | `drive_turn` archived the session (the lib already compacted + retried and it still overflowed); requeue inflight so the message redelivers into a fresh session that fits — no status park | | `Err(RateLimited)` | sleep `HIVE_RATE_LIMIT_SLEEP_SECS` (default 300), requeue inflight, status back to `online` | | `Err(AuthFailed)` | emit `needs_login_idle` sentinel, requeue inflight, park in `wait_for_login` | +| `Err(SessionNotFound)` | resume + create self-heal both missed ("shouldn't happen"); requeue inflight so the next turn creates fresh — no status park, message not dropped | | `Err(Failed(err))` | route `[system] \`\` claude turn failed:\n` to `` via `send_to_parent` | After the outcome handler, the stats sink records a row and the diff --git a/hive-ag3nt/src/bin/hive.rs b/hive-ag3nt/src/bin/hive.rs index 6e710138..0dc652ce 100644 --- a/hive-ag3nt/src/bin/hive.rs +++ b/hive-ag3nt/src/bin/hive.rs @@ -627,6 +627,13 @@ async fn handle_turn( tracing::warn!("prompt-too-long; session archived, requeueing message for a fresh turn"); S::requeue_inflight(socket).await; } + if matches!(outcome, Err(turn::TurnError::SessionNotFound)) { + // "Shouldn't happen": resume missed and the lib's create self-heal + // didn't resolve it. Requeue rather than ack-and-drop so the wake + // message isn't silently lost; the next turn creates the session fresh. + tracing::warn!("session-not-found; 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; } diff --git a/hive-ag3nt/src/serve_common.rs b/hive-ag3nt/src/serve_common.rs index e23bd4c0..4ea4b570 100644 --- a/hive-ag3nt/src/serve_common.rs +++ b/hive-ag3nt/src/serve_common.rs @@ -105,6 +105,7 @@ pub fn build_row(args: TurnRowArgs<'_>) -> TurnStatRow { Err(TurnError::PromptTooLong) => ("prompt_too_long", None), Err(TurnError::RateLimited) => ("rate_limited", None), Err(TurnError::AuthFailed) => ("auth_failed", None), + Err(TurnError::SessionNotFound) => ("session_not_found", None), Err(TurnError::Failed(e)) => ("failed", Some(format!("{e:#}"))), }; let wake_from = if wake_from.starts_with("bash-task-") { diff --git a/hive-ag3nt/src/turn.rs b/hive-ag3nt/src/turn.rs index 78d6a329..bbaf5708 100644 --- a/hive-ag3nt/src/turn.rs +++ b/hive-ag3nt/src/turn.rs @@ -170,6 +170,12 @@ pub enum TurnError { /// into `needs_login_idle` and stop driving turns until the /// operator re-auths via the per-agent web UI. AuthFailed, + /// `--resume ` missed AND the lib's create self-heal also failed to + /// resolve the session — "shouldn't happen" (a resume-miss is normally + /// self-healed inside [`InfiniteSession::attempt`]). Rather than ack + drop + /// the wake message, the serve loop requeues it so the next turn retries; + /// no status park. + SessionNotFound, /// A hard failure with no recovery — the serve loop escalates it to the /// parent (`send_to_parent`). Failed(anyhow::Error), @@ -342,7 +348,11 @@ pub async fn drive_turn( // 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() && outcome.is_ok() { + // `is_ok()` first: `take_compact()` clears the flag, so it must only fire + // when the compaction will actually run. On an unhealthy turn + // (rate-limited / auth-failed / failed) the flag is left set for the next + // turn or the idle `run_pending_compact` to service — not silently eaten. + if outcome.is_ok() && bus.take_compact() { bus.emit(LiveEvent::Note { text: "operator: /compact — running at turn end".into(), }); @@ -426,6 +436,13 @@ pub fn emit_turn_end(bus: &Bus, outcome: &TurnOutcome) { }); tracing::warn!("turn auth-failed (401)"); } + Err(TurnError::SessionNotFound) => { + bus.emit(LiveEvent::TurnEnd { + ok: false, + note: Some("session resume + create both missed — requeueing".into()), + }); + tracing::warn!("turn session-not-found; requeueing message"); + } Err(TurnError::Failed(e)) => { let note = format!("{e:#}"); bus.emit(LiveEvent::TurnEnd { @@ -537,9 +554,7 @@ fn error_to_turn(err: hive_claude::Error) -> TurnOutcome { 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), + Error::SessionNotFound => Err(TurnError::SessionNotFound), other => Err(TurnError::Failed(other.into())), } } @@ -591,10 +606,20 @@ impl Sink for BusSink<'_> { /// watermark). Skips a degenerate turn that parsed nothing so it doesn't reset /// the badges to zero. fn apply_telemetry(bus: &Bus, telemetry: &hive_claude::Telemetry) { - if telemetry.context.context_tokens() == 0 && telemetry.cost.context_tokens() == 0 { + // On a degenerate turn that emitted a `result` but no `assistant` event, + // the per-inference `context` stays zero while `cost` (cumulative) is not. + // Fall back to `cost` as the ctx proxy so the ctx badge + auto-reset + // watermark don't go stale-to-zero. Only a turn that parsed nothing at all + // (both zero) is skipped. + let ctx = if telemetry.context.context_tokens() == 0 { + telemetry.cost + } else { + telemetry.context + }; + if ctx.context_tokens() == 0 { return; } - bus.record_turn_usage(telemetry.context, telemetry.cost); + bus.record_turn_usage(ctx, telemetry.cost); bus.set_resolved_model(telemetry.model.clone()); if let Some(window) = telemetry.context_window { bus.set_api_context_window(window);