fix(agent): don't eat /compact flag on failed turn; restore ctx fallback; requeue on SessionNotFound

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] \`<qualified-label>\` claude turn failed:\n<err>` to `<parent>` via `send_to_parent` |
 
 After the outcome handler, the stats sink records a row and the

hive-ag3nt/src/bin/hive.rs

@@ -627,6 +627,13 @@ async fn handle_turn<S: Surface>(
         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;
     }

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-") {

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 <title>` 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);