diff --git a/src/openhuman/memory_queue/store.rs b/src/openhuman/memory_queue/store.rs index 83c2db1cb..f03f8f2cd 100644 --- a/src/openhuman/memory_queue/store.rs +++ b/src/openhuman/memory_queue/store.rs @@ -537,6 +537,25 @@ pub fn count_by_status(config: &Config, status: JobStatus) -> Result { }) } +/// #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 { + 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 { 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 diff --git a/src/openhuman/memory_tree/tree/rpc.rs b/src/openhuman/memory_tree/tree/rpc.rs index ea1f0b473..03f566f22 100644 --- a/src/openhuman/memory_tree/tree/rpc.rs +++ b/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 { + 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, °raded, ); @@ -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) { @@ -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 = 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, °raded). + // 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"); } diff --git a/tests/memory_tree_sync_deep_raw_coverage_e2e.rs b/tests/memory_tree_sync_deep_raw_coverage_e2e.rs index 48ba51b7a..4b3b3520f 100644 --- a/tests/memory_tree_sync_deep_raw_coverage_e2e.rs +++ b/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)