fix(memory): route only unrecoverable job failures to error status (#3365) (#3630)

Co-authored-by: Claude <noreply@anthropic.com>
Co-authored-by: Steven Enamakel <enamakel@tinyhumans.ai>

src/openhuman/memory_queue/store.rs

@@ -537,6 +537,25 @@ pub fn count_by_status(config: &Config, status: JobStatus) -> Result<u64> {
     })
 }
 
+/// #3365: count terminally-`failed` jobs whose typed class is `unrecoverable`
+/// — the ones `requeue_transient_failed` deliberately leaves parked (budget /
+/// auth / dim-mismatch) because retrying can't help and the user must act. The
+/// status surface routes ONLY these to a hard `error`; transient failures (or
+/// untyped/NULL rows) are auto-requeued and self-heal, so they stay `degraded`.
+/// The `failure_class = 'unrecoverable'` predicate is the exact complement of
+/// the requeue gate's `IS NULL OR != 'unrecoverable'`, so the two never drift.
+pub fn count_failed_unrecoverable(config: &Config) -> Result<u64> {
+    with_connection(config, |conn| {
+        let n: i64 = conn.query_row(
+            "SELECT COUNT(*) FROM mem_tree_jobs \
+              WHERE status = 'failed' AND failure_class = 'unrecoverable'",
+            [],
+            |r| r.get(0),
+        )?;
+        Ok(n.max(0) as u64)
+    })
+}
+
 /// Total count regardless of status — handy for assertions.
 pub fn count_total(config: &Config) -> Result<u64> {
     with_connection(config, |conn| {
@@ -740,6 +759,51 @@ mod tests {
         assert_eq!(row.last_error.as_deref(), Some("Insufficient budget"));
     }
 
+    /// #3365: `count_failed_unrecoverable` counts ONLY terminally-failed jobs
+    /// whose class is `unrecoverable`. Transient-class failures (terminated by
+    /// exhausting `max_attempts`) and untyped/NULL-class failures self-heal via
+    /// `requeue_transient_failed`, so they're excluded — the complement of the
+    /// requeue gate. The status surface uses this to route only the former to
+    /// `error`.
+    #[test]
+    fn count_failed_unrecoverable_excludes_transient_and_untyped() {
+        use crate::openhuman::memory_tree::health::{FailureCode, PipelineFailure};
+        let (_tmp, cfg) = test_config();
+
+        // Helper: enqueue a job, claim it, then mark it failed with `failure`.
+        let fail_one = |chunk: &str, max_attempts: u32, failure: Option<&PipelineFailure>| {
+            let mut nj = NewJob::extract_chunk(&ExtractChunkPayload {
+                chunk_id: chunk.into(),
+            })
+            .unwrap();
+            nj.max_attempts = Some(max_attempts);
+            enqueue(&cfg, &nj).unwrap().expect("inserted");
+            let claimed = claim_next(&cfg, DEFAULT_LOCK_DURATION_MS).unwrap().unwrap();
+            mark_failed_typed(&cfg, &claimed, "boom", failure).unwrap();
+        };
+
+        // Unrecoverable ⇒ terminal on first attempt, class persisted ⇒ counted.
+        let unrec = PipelineFailure::new(FailureCode::BudgetExhausted);
+        fail_one("c-unrec", 5, Some(&unrec));
+        // Transient ⇒ terminal only because max_attempts=1 is exhausted; class
+        // persisted as 'transient' ⇒ NOT counted (it will be auto-requeued).
+        let trans = PipelineFailure::new(FailureCode::Transient);
+        fail_one("c-trans", 1, Some(&trans));
+        // Untyped terminal (NULL class) ⇒ NOT counted.
+        fail_one("c-null", 1, None);
+
+        assert_eq!(
+            count_by_status(&cfg, JobStatus::Failed).unwrap(),
+            3,
+            "all three jobs are terminally failed"
+        );
+        assert_eq!(
+            count_failed_unrecoverable(&cfg).unwrap(),
+            1,
+            "only the unrecoverable one is counted"
+        );
+    }
+
     /// T012: a **transient** typed failure keeps the existing
     /// attempts-bounded retry path — the job bounces back to `ready` with a
     /// future `available_at_ms` and does NOT set the typed columns (they are

src/openhuman/memory_tree/tree/rpc.rs

@@ -368,21 +368,29 @@ pub async fn pipeline_status_rpc(
             msg
         })??;
 
-    // Job counters — three parallel-safe blocking calls.
+    // Job counters — parallel-safe blocking calls. `failed_unrecoverable` is the
+    // #3365 left-right split: of the failed jobs, how many are the hard,
+    // user-actionable kind (`failure_class = 'unrecoverable'`) vs transient ones
+    // that self-heal via auto-requeue. Only the former escalates to `error`.
     let cfg_for_jobs = config.clone();
-    let pipeline_jobs =
-        tokio::task::spawn_blocking(move || -> Result<PipelineJobCounts, String> {
+    let (pipeline_jobs, failed_unrecoverable) =
+        tokio::task::spawn_blocking(move || -> Result<(PipelineJobCounts, u64), String> {
             let ready = queue_store::count_by_status(&cfg_for_jobs, JobStatus::Ready)
                 .map_err(|e| format!("count_by_status(ready): {e:#}"))?;
             let running = queue_store::count_by_status(&cfg_for_jobs, JobStatus::Running)
                 .map_err(|e| format!("count_by_status(running): {e:#}"))?;
             let failed = queue_store::count_by_status(&cfg_for_jobs, JobStatus::Failed)
                 .map_err(|e| format!("count_by_status(failed): {e:#}"))?;
-            Ok(PipelineJobCounts {
-                ready,
-                running,
-                failed,
-            })
+            let failed_unrecoverable = queue_store::count_failed_unrecoverable(&cfg_for_jobs)
+                .map_err(|e| format!("count_failed_unrecoverable: {e:#}"))?;
+            Ok((
+                PipelineJobCounts {
+                    ready,
+                    running,
+                    failed,
+                },
+                failed_unrecoverable,
+            ))
         })
         .await
         .map_err(|e| {
@@ -411,7 +419,11 @@ pub async fn pipeline_status_rpc(
 
     // #002: read the process-global degradation snapshot (set by the embed /
     // extract stages) so a half-working sync surfaces as `degraded` with a
-    // cause rather than a misleading `running`.
+    // cause rather than a misleading `running`. The structure-degraded latch is
+    // a liveness signal ("the extraction model is timing out") kept honest at
+    // its source in `extract::llm` — it self-clears on the next *completed*
+    // extraction (#3365), so the status surface never consults the unrelated
+    // `extraction_coverage` metric to second-guess it here.
     let degraded = crate::openhuman::memory_tree::health::current_degraded_state();
 
     let (status, reason) = derive_pipeline_status(
@@ -419,6 +431,7 @@ pub async fn pipeline_status_rpc(
         config.scheduler_gate.mode,
         is_syncing,
         pipeline_jobs.failed,
+        failed_unrecoverable,
         total_chunks,
         &degraded,
     );
@@ -429,7 +442,10 @@ pub async fn pipeline_status_rpc(
     // to `None` rather than failing the polled status RPC.
     //   - first_blocking_cause (FR-004): the most-recent failed job's typed
     //     reason, surfaced verbatim by the UI.
-    //   - extraction_coverage (FR-010/US5): fraction of chunks with structure.
+    //   - extraction_coverage (FR-010/US5): fraction of chunks with structure,
+    //     surfaced as its own display metric — deliberately NOT folded into the
+    //     status pill (#3365: coverage is a cumulative measure, unrelated to the
+    //     live structure-degraded liveness signal).
     //     `None` (not `0.0`) on a read error, so a broken measurement path is
     //     never mistaken for a genuine 0% extraction rate.
     let (latest_failure, extraction_coverage) = {
@@ -634,6 +650,7 @@ fn derive_pipeline_status(
     mode: crate::openhuman::config::SchedulerGateMode,
     is_syncing: bool,
     failed: u64,
+    failed_unrecoverable: u64,
     total_chunks: u64,
     degraded: &crate::openhuman::memory_tree::health::DegradedState,
 ) -> (String, Option<String>) {
@@ -643,28 +660,44 @@ fn derive_pipeline_status(
             Some(format!("scheduler gate mode = {}", mode.as_str())),
         );
     }
-    if failed > 0 {
+    // #3365: split the failed bucket by class. Only an UNRECOVERABLE failure
+    // (budget / auth / dim-mismatch) is a hard `error` the user must act on —
+    // it stays parked and can't self-heal. Transient failures are auto-requeued
+    // by `requeue_transient_failed`, so they must NOT escalate to `error`; they
+    // fall through to `degraded` ("failed, retrying") below. This fixes the prior
+    // `failed > 0 → error` that flashed a scary error for a job about to retry.
+    if failed_unrecoverable > 0 {
         return (
             "error".to_string(),
-            Some(format!("{failed} failed job(s) in pipeline")),
+            Some(format!(
+                "{failed_unrecoverable} unrecoverable failure(s) need action"
+            )),
         );
     }
     // #002 (FR-005): "degraded" sits below error but above syncing/running —
-    // the pipeline is making progress, but recall/structure is reduced and the
-    // user should be told why. Beats syncing/running so a half-working sync
-    // isn't reported as plain "running"/"syncing".
+    // the pipeline is making progress, but recall/structure is reduced (or some
+    // jobs failed transiently and are retrying) and the user should be told why.
+    // Beats syncing/running so a half-working sync isn't reported as plain
+    // "running"/"syncing".
     //
     // Only fires when there are chunks: degraded recall/structure is only
     // meaningful when there's actual content affected. An empty workspace with
     // a misconfigured embedder should show "idle" (nothing to recall) rather
     // than "degraded" (recall is broken for existing content).
-    if degraded.is_degraded() && total_chunks > 0 {
-        let mut parts = Vec::new();
+    //
+    // `failed` here is transient-only — any unrecoverable failure returned
+    // `error` above, so a non-zero `failed` at this point means jobs that will
+    // be auto-requeued.
+    if (degraded.is_degraded() || failed > 0) && total_chunks > 0 {
+        let mut parts: Vec<String> = Vec::new();
         if degraded.semantic_recall {
-            parts.push("semantic recall disabled");
+            parts.push("semantic recall disabled".to_string());
         }
         if degraded.structure {
-            parts.push("wiki structure incomplete");
+            parts.push("wiki structure incomplete".to_string());
+        }
+        if failed > 0 {
+            parts.push(format!("{failed} job(s) failed, retrying"));
         }
         return ("degraded".to_string(), Some(parts.join("; ")));
     }
@@ -1067,23 +1100,43 @@ mod tests {
             cause: Some(PipelineFailure::new(FailureCode::ExtractionTimeout)),
         };
 
+        // Args: (is_paused, mode, is_syncing, failed, failed_unrecoverable,
+        //        total_chunks, &degraded).
+
         // paused beats everything else (even degradation)
-        let (s, reason) =
-            derive_pipeline_status(true, SchedulerGateMode::Off, true, 5, 100, &recall_degraded);
+        let (s, reason) = derive_pipeline_status(
+            true,
+            SchedulerGateMode::Off,
+            true,
+            5,
+            5,
+            100,
+            &recall_degraded,
+        );
         assert_eq!(s, "paused");
         assert!(reason.unwrap().contains("off"));
 
-        // error beats degraded / syncing / running / idle
+        // error beats degraded / syncing / running / idle — but ONLY for
+        // unrecoverable failures (#3365).
         let (s, reason) = derive_pipeline_status(
             false,
             SchedulerGateMode::Auto,
             true,
             2,
+            2, // both failures unrecoverable
             100,
             &recall_degraded,
         );
         assert_eq!(s, "error");
-        assert!(reason.unwrap().contains("2 failed"));
+        assert!(reason.unwrap().contains("unrecoverable"));
+
+        // #3365: transient-only failures (failed > 0, none unrecoverable) do NOT
+        // escalate to error — they self-heal via auto-requeue, so they surface
+        // as `degraded` ("retrying"), beating syncing/running.
+        let (s, reason) =
+            derive_pipeline_status(false, SchedulerGateMode::Auto, true, 3, 0, 100, &healthy);
+        assert_eq!(s, "degraded", "transient failures must not read as error");
+        assert!(reason.unwrap().contains("3 job(s) failed, retrying"));
 
         // #002: degraded beats syncing / running / idle (but loses to paused/error)
         let (s, reason) = derive_pipeline_status(
@@ -1091,6 +1144,7 @@ mod tests {
             SchedulerGateMode::Auto,
             true, // syncing
             0,
+            0,
             100,
             &recall_degraded,
         );
@@ -1102,6 +1156,7 @@ mod tests {
             SchedulerGateMode::Auto,
             false,
             0,
+            0,
             100,
             &structure_degraded,
         );
@@ -1110,17 +1165,19 @@ mod tests {
 
         // syncing beats running / idle (when healthy)
         let (s, reason) =
-            derive_pipeline_status(false, SchedulerGateMode::Auto, true, 0, 100, &healthy);
+            derive_pipeline_status(false, SchedulerGateMode::Auto, true, 0, 0, 100, &healthy);
         assert_eq!(s, "syncing");
         assert!(reason.is_none());
 
         // running when chunks exist but nothing in flight
         let (s, _) =
-            derive_pipeline_status(false, SchedulerGateMode::Auto, false, 0, 100, &healthy);
+            derive_pipeline_status(false, SchedulerGateMode::Auto, false, 0, 0, 100, &healthy);
         assert_eq!(s, "running");
 
-        // idle when the store is empty and nothing is in flight
-        let (s, _) = derive_pipeline_status(false, SchedulerGateMode::Auto, false, 0, 0, &healthy);
+        // idle when the store is empty and nothing is in flight (transient
+        // failures with no content don't manufacture a `degraded`).
+        let (s, _) =
+            derive_pipeline_status(false, SchedulerGateMode::Auto, false, 2, 0, 0, &healthy);
         assert_eq!(s, "idle");
     }
 

tests/memory_tree_sync_deep_raw_coverage_e2e.rs

@@ -499,6 +499,8 @@ async fn memory_tree_rpc_status_set_enabled_backfill_and_ingest_errors() {
         .expect("build failed job"),
     )
     .expect("enqueue failed job");
+    // #3365: an untyped/transient failed job (no `failure_class`) self-heals via
+    // requeue, so it surfaces as `degraded` ("retrying"), not a hard `error`.
     with_connection(&cfg, |conn| {
         conn.execute(
             "UPDATE mem_tree_jobs
@@ -509,10 +511,30 @@ async fn memory_tree_rpc_status_set_enabled_backfill_and_ingest_errors() {
         )?;
         Ok(())
     })
-    .expect("mark failed");
+    .expect("mark failed (untyped)");
+    let degraded = pipeline_status_rpc(&cfg)
+        .await
+        .expect("degraded status")
+        .value;
+    assert_eq!(degraded.status, "degraded");
+    assert!(degraded.reason.unwrap().contains("retrying"));
+
+    // #3365: an UNRECOVERABLE failed job (budget / auth / dim-mismatch) stays
+    // parked and is user-actionable, so it escalates to `error`.
+    with_connection(&cfg, |conn| {
+        conn.execute(
+            "UPDATE mem_tree_jobs
+                SET failure_class = 'unrecoverable'
+              WHERE kind = 'extract_chunk'
+                AND payload_json LIKE '%round18-failed%'",
+            [],
+        )?;
+        Ok(())
+    })
+    .expect("mark unrecoverable");
     let errored = pipeline_status_rpc(&cfg).await.expect("error status").value;
     assert_eq!(errored.status, "error");
-    assert!(errored.reason.unwrap().contains("failed job"));
+    assert!(errored.reason.unwrap().contains("unrecoverable"));
 
     cfg.scheduler_gate.mode = SchedulerGateMode::Off;
     let paused = pipeline_status_rpc(&cfg)