From 0aed700f9c9c4ea905148074e702c3d8709decd8 Mon Sep 17 00:00:00 2001 From: Valeriy Selitskiy <239034+iamwavecut@users.noreply.github.com> Date: Fri, 6 Nov 2026 23:19:12 +0100 Subject: [PATCH] Keep memory runs and alerts alive when an LLM backend is unreachable While ninfer-3090 was down after the 2026-11-05 farm reboot, Discovery failed every job with result.error {UPSTREAM_ERROR, name resolution}. The bot read only job.error, so every failure became the generic "dialog job failed": it was not retryable, the breaker never opened, the routing admin report stayed silent for 30 h, and each memory run spent all five attempts within half an hour (7,897 runs retired). - Decode Discovery's result.error and report it as "CODE: message". UPSTREAM_ERROR, SERVICE_NOT_FOUND, SERVICE_DISABLED and ENDPOINT_NOT_FOUND become AifarmClientError::BackendUnavailable, classified as ProviderUnavailable like Transport. - Release a memory run instead of failing it when extraction fails because no LLM backend could take the call. - Collapse a deferred memory pipeline to one followup job delayed by the retry interval, so an outage is probed every 15 s instead of spinning every worker on claims. --- crates/openplotva-app/src/embedder.rs | 1 + crates/openplotva-app/src/memory_runtime.rs | 326 ++++++++++++++++++- crates/openplotva-app/src/routed_attempts.rs | 56 +++- crates/openplotva-llm/src/aifarm.rs | 150 ++++++++- crates/openplotva-llm/src/retry.rs | 25 +- 5 files changed, 541 insertions(+), 17 deletions(-) diff --git a/crates/openplotva-app/src/embedder.rs b/crates/openplotva-app/src/embedder.rs index 8790e459..90a56d20 100644 --- a/crates/openplotva-app/src/embedder.rs +++ b/crates/openplotva-app/src/embedder.rs @@ -825,6 +825,7 @@ mod tests { DiscoveryJob { state: "JOB_STATE_SUCCEEDED".to_owned(), result: Some(DiscoveryJobResult { + error: None, response: Some(DiscoveryJobResponse { status_code, body: general_purpose::STANDARD.encode(body_json), diff --git a/crates/openplotva-app/src/memory_runtime.rs b/crates/openplotva-app/src/memory_runtime.rs index c9363c13..633103e0 100644 --- a/crates/openplotva-app/src/memory_runtime.rs +++ b/crates/openplotva-app/src/memory_runtime.rs @@ -26,7 +26,8 @@ use openplotva_storage::{ DialogMemoryChatMeta, PgEmbeddingVector, PostgresMemoryStore, StorageError, }; use openplotva_taskman::{ - InMemoryTaskQueue, JobType, MEMORY_CONSOLIDATION_QUEUE_NAME, new_memory_consolidation_job_at, + InMemoryTaskQueue, JobType, MEMORY_CONSOLIDATION_QUEUE_NAME, TaskQueueSchedule, + new_memory_consolidation_job_at, }; use serde::{Deserialize, Serialize}; use serde_json::Value; @@ -36,15 +37,17 @@ use time::OffsetDateTime; use crate::embedder::DiscoveryEmbedderClient; use crate::routed_attempts::{ RoutedAttempt, RoutedAttemptRunError, RoutedAttemptWalker, RoutedRequestContext, - RoutedRoutingError, + RoutedRoutingError, RoutedRoutingErrorKind, }; use crate::runtime_gemini_cache::resolve_google_ai_key; pub const EMBEDDER_DEFAULT_TIMEOUT: Duration = Duration::from_secs(60); pub const MEMORY_RETRIEVAL_EMBEDDING_TIMEOUT: Duration = Duration::from_secs(15); pub const MEMORY_CONSOLIDATION_JOB_POLL_INTERVAL: Duration = Duration::from_secs(1); -/// How long a worker waits before the next job after the embedder gate deferred one. -pub const MEMORY_CONSOLIDATION_EMBEDDER_RETRY_INTERVAL: Duration = Duration::from_secs(15); +/// How long a deferred pipeline waits before it probes its backends again: the worker +/// that deferred pauses this long, and the one job the pipeline keeps queued becomes +/// available only then. +pub const MEMORY_CONSOLIDATION_BACKEND_RETRY_INTERVAL: Duration = Duration::from_secs(15); pub const MEMORY_CONSOLIDATION_JOB_WORKER_PREFIX: &str = "memory-consolidation-worker"; pub const MEMORY_EXTRACTION_WORKFLOW_KEY: &str = "memory_extraction"; pub const MEMORY_SUBJECT_MERGE_WORKFLOW_KEY: &str = "memory_subject_merge"; @@ -934,6 +937,39 @@ where pub fn is_embedder_unavailable(&self) -> bool { matches!(self, Self::EmbedderUnavailable { .. }) } + + /// Whether the extractor failed because no LLM backend could take the call: + /// unreachable, unregistered, out of capacity or behind open breakers. The run + /// is not at fault, so it is released without spending an attempt. + #[must_use] + pub fn is_llm_backend_unavailable(&self) -> bool { + match self { + Self::Extract { source } => llm_backend_unavailable(source), + _ => false, + } + } +} + +fn llm_backend_unavailable(error: &(dyn StdError + 'static)) -> bool { + let mut current = Some(error); + while let Some(error) = current { + if let Some(RoutedMemoryExtractorError::Routing(routing)) = + error.downcast_ref::() + { + return matches!( + routing.kind(), + RoutedRoutingErrorKind::RouteUnavailable + | RoutedRoutingErrorKind::NoCandidates + | RoutedRoutingErrorKind::Exhausted + | RoutedRoutingErrorKind::CapacityWaitTimeout + ); + } + current = error.source(); + } + matches!( + retryable_reason(error), + Some(FailureReason::ProviderUnavailable | FailureReason::CapacityUnavailable) + ) } /// Memory run processing failure. @@ -1263,6 +1299,23 @@ where ..MemoryRunProcessReport::default() }) } + // No LLM backend could take the call. The run is not at fault, and failing it + // would spend all its attempts within minutes of an outage. + Err(error) if error.is_llm_backend_unavailable() => { + tracing::warn!( + run_id, + error = %error, + "memory LLM backend unavailable; releasing memory run for retry" + ); + store + .release_run(run_id) + .await + .map_err(|source| MemoryRunProcessError::Store { source })?; + Ok(MemoryRunProcessReport { + deferred: true, + ..MemoryRunProcessReport::default() + }) + } Err(error) => { let text = error.to_string(); store @@ -1622,12 +1675,23 @@ where Err(error) => report.status_error = Some(error.to_string()), } // Pipeline jobs exist only while ticks keep scheduling them, so a tick - // deferred by an embedder outage keeps one queued for the retry. - if processed.deferred { + // deferred by a backend outage keeps exactly one job queued, delayed by the + // retry interval: the pipeline probes the backend instead of every worker + // spinning on claims. A productive probe ramps it back up. + if processed.deferred + && queue.queue_depth( + MEMORY_CONSOLIDATION_QUEUE_NAME, + openplotva_taskman::LOWEST_PRIORITY, + ) == 0 + { report.scheduled_followup = true; - report.followup_job_id = Some(queue.assign( + report.followup_job_id = Some(queue.assign_with_schedule( MEMORY_CONSOLIDATION_QUEUE_NAME, new_memory_consolidation_job_at(now), + TaskQueueSchedule { + available_at: Some(now + MEMORY_CONSOLIDATION_BACKEND_RETRY_INTERVAL), + ..TaskQueueSchedule::default() + }, )); } if processed.processed { @@ -1712,7 +1776,7 @@ where ).await; tracing::debug!(?tick, "memory-consolidation taskman worker tick"); pause = if tick.deferred { - MEMORY_CONSOLIDATION_EMBEDDER_RETRY_INTERVAL + MEMORY_CONSOLIDATION_BACKEND_RETRY_INTERVAL } else { cfg.interval }; @@ -4915,6 +4979,7 @@ mod tests { assert_eq!(merger.calls.load(std::sync::atomic::Ordering::SeqCst), 2); } + use openplotva_llm::aifarm::AifarmClientError; use openplotva_storage::pg_embedding_vector; use std::sync::Mutex; @@ -5316,6 +5381,59 @@ mod tests { } } + struct FailingExtractor { + error: fn() -> RoutedMemoryExtractorError, + } + + impl MemoryExtractor for FailingExtractor { + type Error = RoutedMemoryExtractorError; + + fn extract<'a>( + &'a self, + _input: &'a ExtractInput, + ) -> MemoryExtractorFuture<'a, Self::Error> { + Box::pin(async move { Err((self.error)()) }) + } + } + + fn aifarm_completion_error(error: AifarmClientError) -> RoutedMemoryExtractorError { + RoutedMemoryExtractorError::Aifarm(AifarmMemoryExtractorError::Completion { + source: Box::new(error), + }) + } + + fn gpu2_unreachable() -> RoutedMemoryExtractorError { + aifarm_completion_error(AifarmClientError::BackendUnavailable( + "UPSTREAM_ERROR: [Errno -3] Temporary failure in name resolution".to_owned(), + )) + } + + fn gpu2_timed_out() -> RoutedMemoryExtractorError { + aifarm_completion_error(AifarmClientError::Upstream( + "UPSTREAM_TIMEOUT: read timed out".to_owned(), + )) + } + + fn one_message_run_store(run_id: i64) -> FakeMemoryWriteStore { + FakeMemoryWriteStore { + run: Mutex::new(Some(openplotva_memory::Run { + id: run_id, + chat_id: 42, + prompt_version: openplotva_memory::PROMPT_VERSION.to_owned(), + message_count: 1, + ..openplotva_memory::Run::default() + })), + messages: Mutex::new(vec![openplotva_memory::Message { + entry_id: "msg:1".to_owned(), + message_id: 1, + text: "Alice likes Rust".to_owned(), + occurred_at: OffsetDateTime::UNIX_EPOCH + time::Duration::seconds(1), + ..openplotva_memory::Message::default() + }]), + ..FakeMemoryWriteStore::default() + } + } + struct ResolvingExtractor { output: ExtractOutput, plan: openplotva_memory::ResolutionPlan, @@ -6378,6 +6496,124 @@ mod tests { ); } + #[test] + fn llm_backend_unavailable_covers_unreachable_backends_and_closed_routes() { + type WriteError = MemoryExtractionWriteError; + let extract = |source| WriteError::Extract { source }; + + for source in [ + gpu2_unreachable(), + aifarm_completion_error(AifarmClientError::Transport( + "error sending request for url".to_owned(), + )), + aifarm_completion_error(AifarmClientError::Upstream( + "upstream returned status 503: The model is still loading. Retry shortly." + .to_owned(), + )), + aifarm_completion_error(AifarmClientError::Upstream( + "service capacity unavailable".to_owned(), + )), + RoutedMemoryExtractorError::Routing(RoutedRoutingError::new( + RoutedRoutingErrorKind::NoCandidates, + "workflow route has no eligible candidates", + )), + RoutedMemoryExtractorError::Routing(RoutedRoutingError::new( + RoutedRoutingErrorKind::Exhausted, + "routed attempts exhausted", + )), + ] { + let error = extract(source); + assert!(error.is_llm_backend_unavailable(), "{error}"); + } + + for source in [ + gpu2_timed_out(), + aifarm_completion_error(AifarmClientError::Upstream("dialog job failed".to_owned())), + RoutedMemoryExtractorError::Aifarm(AifarmMemoryExtractorError::Response( + "memory extractor returned empty content".to_owned(), + )), + RoutedMemoryExtractorError::Routing(RoutedRoutingError::new( + RoutedRoutingErrorKind::DeadlineExceeded, + "routed attempt exceeded the caller deadline", + )), + ] { + let error = extract(source); + assert!(!error.is_llm_backend_unavailable(), "{error}"); + } + assert!( + !WriteError::StoreCards { + source: TestStoreError + } + .is_llm_backend_unavailable(), + "store failures are never an LLM outage" + ); + } + + #[tokio::test] + async fn process_next_memory_run_releases_run_when_llm_backend_is_unreachable() { + let store = one_message_run_store(12); + let extractor = FailingExtractor { + error: gpu2_unreachable, + }; + + let report = process_next_memory_run( + &extractor, + &store, + Option::<&FakeEmbedder>::None, + MemoryRunProcessConfig { + max_messages_per_run: 10, + episode_model: "model".to_owned(), + ..MemoryRunProcessConfig::default() + }, + ) + .await + .expect("an LLM outage is not a run failure"); + + assert!( + report.deferred, + "the pipeline backs off until the backend returns" + ); + assert_eq!(store.released.lock().expect("released").as_slice(), &[12]); + assert!( + store.failed.lock().expect("failed").is_empty(), + "an LLM outage must not burn the run's retry budget" + ); + } + + #[tokio::test] + async fn process_next_memory_run_spends_an_attempt_when_the_backend_answered_badly() { + let store = one_message_run_store(13); + let extractor = FailingExtractor { + error: gpu2_timed_out, + }; + + let result = process_next_memory_run( + &extractor, + &store, + Option::<&FakeEmbedder>::None, + MemoryRunProcessConfig { + max_messages_per_run: 10, + episode_model: "model".to_owned(), + ..MemoryRunProcessConfig::default() + }, + ) + .await; + + assert!(matches!( + result, + Err(MemoryRunProcessError::Process { run_id: 13, .. }) + )); + assert!(store.released.lock().expect("released").is_empty()); + assert_eq!( + store.failed.lock().expect("failed").as_slice(), + &[( + 13, + "extract memory batch: generate memory extraction: UPSTREAM_TIMEOUT: read timed out" + .to_owned() + )] + ); + } + #[tokio::test] async fn process_next_memory_run_selects_by_token_budget_and_adds_context() { let run = openplotva_memory::Run { @@ -6904,6 +7140,80 @@ mod tests { queue.record(2).expect("followup job").status, openplotva_taskman::JobStatus::Pending ); + assert!( + queue + .dequeue( + openplotva_taskman::MEMORY_CONSOLIDATION_QUEUE_NAME, + "memory-consolidation-worker-1", + now, + ) + .is_none(), + "the followup waits out the retry interval" + ); + assert!( + queue + .dequeue( + openplotva_taskman::MEMORY_CONSOLIDATION_QUEUE_NAME, + "memory-consolidation-worker-1", + now + MEMORY_CONSOLIDATION_BACKEND_RETRY_INTERVAL, + ) + .is_some() + ); + } + + #[tokio::test] + async fn memory_taskman_pipeline_collapses_to_one_probe_during_an_llm_outage() { + let now = utc_test_time(3, 17); + let queue = openplotva_taskman::InMemoryTaskQueue::new(); + for _ in 0..3 { + queue.assign( + openplotva_taskman::MEMORY_CONSOLIDATION_QUEUE_NAME, + openplotva_taskman::new_memory_consolidation_job_at(now), + ); + } + let extractor = FailingExtractor { + error: gpu2_unreachable, + }; + + let mut followups = Vec::new(); + for worker in 0..3 { + let store = one_message_run_store(100 + worker); + let report = process_memory_consolidation_taskman_job_once_at( + &queue, + &extractor, + &store, + Option::<&FakeEmbedder>::None, + MemoryRunProcessConfig { + max_messages_per_run: 10, + episode_model: "model".to_owned(), + ..MemoryRunProcessConfig::default() + }, + &format!("memory-consolidation-worker-{worker}"), + 4, + now, + ) + .await; + assert!(report.deferred); + assert!( + !report.failed, + "an outage completes the job, it does not fail it" + ); + assert_eq!(store.released.lock().expect("released").len(), 1); + followups.extend(report.followup_job_id); + } + + assert_eq!( + followups.len(), + 1, + "only the last deferral, with nothing else queued, keeps a probe" + ); + assert_eq!( + queue.queue_depth( + openplotva_taskman::MEMORY_CONSOLIDATION_QUEUE_NAME, + openplotva_taskman::LOWEST_PRIORITY, + ), + 1 + ); } #[tokio::test] diff --git a/crates/openplotva-app/src/routed_attempts.rs b/crates/openplotva-app/src/routed_attempts.rs index 0eafaebb..862b0d03 100644 --- a/crates/openplotva-app/src/routed_attempts.rs +++ b/crates/openplotva-app/src/routed_attempts.rs @@ -842,7 +842,7 @@ pub struct RoutedRoutingError { } impl RoutedRoutingError { - fn new(kind: RoutedRoutingErrorKind, message: impl Into) -> Self { + pub(crate) fn new(kind: RoutedRoutingErrorKind, message: impl Into) -> Self { Self { kind, message: message.into(), @@ -1651,6 +1651,60 @@ mod tests { ); } + /// Whether six consecutive failures of the only candidate with `error` open its + /// circuit and queue the owner's routing incident report. + async fn outage_reaches_admin_report( + error: fn() -> openplotva_llm::aifarm::AifarmClientError, + ) -> bool { + let reporter = crate::runtime_routing::RoutingEventReporter::new( + crate::runtime_routing::RoutingEventBuffer::new(64), + None, + Some(Arc::new(tokio::sync::Notify::new())), + ); + let walker = + walker_with_breakers(snapshot(json!({}), json!({})), Arc::new(BreakerSet::new())) + .with_reporter(reporter.clone()); + for _ in 0..6 { + let result = walker + .run( + RoutedRequestContext { + workflow_key: "dialog".to_owned(), + ..RoutedRequestContext::default() + }, + move |_attempt| async move { + Err::<&str, openplotva_llm::aifarm::CompletionError>(Box::new(error())) + }, + |error| openplotva_llm::retry::retryable_reason(&**error), + ) + .await; + assert!(result.is_err()); + } + reporter.buffer().routing_events(64).iter().any(|event| { + event.event_type == "circuit_open_exhaustion" + && event.detail["admin_report"]["action"] == "aggregated" + }) + } + + #[tokio::test] + async fn unreachable_discovery_backend_opens_the_circuit_and_reaches_the_owner() { + use openplotva_llm::aifarm::AifarmClientError; + + assert!( + outage_reaches_admin_report(|| AifarmClientError::BackendUnavailable( + "UPSTREAM_ERROR: [Errno -3] Temporary failure in name resolution".to_owned(), + )) + .await + ); + // What the bot saw for 30 h on 2026-11-05/06: a generic failure is not + // retryable, never charges the breaker, and so never reaches the owner. + assert!( + !outage_reaches_admin_report(|| AifarmClientError::Upstream( + "dialog job failed".to_owned() + )) + .await + ); + } + use openplotva_storage::llm_routing::PoolRecord; /// Two chat primaries: model 10 draws from 1-slot pool 1, model 20 is diff --git a/crates/openplotva-llm/src/aifarm.rs b/crates/openplotva-llm/src/aifarm.rs index ba1d70ac..09f1a137 100644 --- a/crates/openplotva-llm/src/aifarm.rs +++ b/crates/openplotva-llm/src/aifarm.rs @@ -601,6 +601,9 @@ pub enum AifarmClientError { /// Upstream non-2xx discovery result. #[error("{0}")] Upstream(String), + /// Discovery could not reach or route to the backend service. + #[error("{0}")] + BackendUnavailable(String), /// Unknown Discovery status. #[error("unknown dialog job status {0:?}")] UnknownStatus(String), @@ -4235,6 +4238,9 @@ pub struct DiscoveryJobResult { /// HTTP response payload. #[serde(default)] pub response: Option, + /// Discovery's own failure, `{code, message}`, when the backend call produced no response. + #[serde(default)] + pub error: Option, } #[derive(Clone, Debug, Default, Deserialize, PartialEq)] @@ -4258,6 +4264,8 @@ pub struct DiscoveryJobStatus { pub status: String, /// Message. pub message: String, + /// Discovery failure code from `result.error`, such as `UPSTREAM_ERROR`. + pub error_code: String, /// Upstream status code. pub status_code: u16, /// Raw decoded body. @@ -5330,10 +5338,18 @@ pub fn discovery_status_from_envelope( envelope: &DiscoveryJobEnvelope, ) -> Result { let job = envelope.resolve_job(); + let (error_code, result_error) = + parse_discovery_result_error(job.result.as_ref().and_then(|result| result.error.as_ref())); + let job_error = parse_job_error(job.error.as_ref()); let mut status = DiscoveryJobStatus { job_id: fallback_string(&job.resolved_id(), fallback_job_id), status: job.resolved_status(), - message: parse_job_error(job.error.as_ref()).trim().to_owned(), + message: if job_error.is_empty() { + result_error + } else { + job_error + }, + error_code, ..DiscoveryJobStatus::default() }; @@ -5347,6 +5363,38 @@ pub fn discovery_status_from_envelope( Ok(status) } +/// Discovery failure codes meaning the backend was never reached: a connection or DNS +/// error (`UPSTREAM_ERROR`) or a service/endpoint that is not registered or is disabled. +/// `UPSTREAM_TIMEOUT` is not one of them: the backend accepted the call and may be busy. +const DISCOVERY_BACKEND_UNAVAILABLE_CODES: &[&str] = &[ + "UPSTREAM_ERROR", + "SERVICE_NOT_FOUND", + "SERVICE_DISABLED", + "ENDPOINT_NOT_FOUND", +]; + +/// Discovery's `result.error` as its code and a `CODE: message` description. +fn parse_discovery_result_error(raw: Option<&Value>) -> (String, String) { + let Some(raw) = raw else { + return (String::new(), String::new()); + }; + let field = |key: &str| { + raw.get(key) + .and_then(Value::as_str) + .map(str::trim) + .unwrap_or_default() + .to_owned() + }; + let (code, message) = (field("code"), field("message")); + let description = match (code.is_empty(), message.is_empty()) { + (false, false) => format!("{code}: {message}"), + (false, true) => code.clone(), + (true, false) => message, + (true, true) => parse_job_error(Some(raw)), + }; + (code, description) +} + fn choose_string(primary: &str, fallback: &str) -> String { if primary.is_empty() { fallback.to_owned() @@ -5377,9 +5425,11 @@ fn discovery_result_from_status( return Ok(Some(status.completion_result())); } if is_failure_status(normalized) { - return Err(Box::new(AifarmClientError::Upstream( - failed_discovery_status_error(status), - ))); + let message = failed_discovery_status_error(status); + if DISCOVERY_BACKEND_UNAVAILABLE_CODES.contains(&status.error_code.as_str()) { + return Err(Box::new(AifarmClientError::BackendUnavailable(message))); + } + return Err(Box::new(AifarmClientError::Upstream(message))); } if status.response.is_some() && (200..300).contains(&status.status_code) { return Ok(Some(status.completion_result())); @@ -10696,6 +10746,7 @@ mod tests { state: "top-state".to_owned(), error: Some(json!(" top-error ")), result: Some(DiscoveryJobResult { + error: None, response: Some(DiscoveryJobResponse { status_code: 202, ..DiscoveryJobResponse::default() @@ -10763,6 +10814,7 @@ mod tests { let body = general_purpose::STANDARD.encode(payload); let result = decode_chat_completion_response(Some(&DiscoveryJobResult { + error: None, response: Some(DiscoveryJobResponse { status_code: 200, body, @@ -10789,6 +10841,7 @@ mod tests { let body = general_purpose::URL_SAFE.encode(payload); let result = decode_chat_completion_response(Some(&DiscoveryJobResult { + error: None, response: Some(DiscoveryJobResponse { status_code: 200, body, @@ -10814,6 +10867,7 @@ mod tests { let body = general_purpose::STANDARD.encode(payload); let result = decode_chat_completion_response(Some(&DiscoveryJobResult { + error: None, response: Some(DiscoveryJobResponse { status_code: 503, body, @@ -10836,6 +10890,7 @@ mod tests { job_id: "job-1".to_owned(), state: "JOB_STATE_FAILED".to_owned(), result: Some(DiscoveryJobResult { + error: None, response: Some(DiscoveryJobResponse { status_code: 503, body, @@ -10856,6 +10911,93 @@ mod tests { Ok(()) } + fn failed_discovery_envelope(code: &str, message: &str) -> DiscoveryJobEnvelope { + let raw = json!({ + "job": { + "job_id": "job-1", + "state": "JOB_STATE_FAILED", + "result": {"response": null, "error": {"code": code, "message": message}}, + } + }); + serde_json::from_value(raw).expect("Discovery failed-job envelope decodes") + } + + #[test] + fn failed_discovery_job_reports_result_error_instead_of_a_generic_failure() + -> Result<(), AifarmDecodeError> { + let envelope = failed_discovery_envelope( + "UPSTREAM_ERROR", + "[Errno -3] Temporary failure in name resolution", + ); + + let status = discovery_status_from_envelope("fallback", &envelope)?; + let update = job_status_from_envelope("fallback", &envelope)?; + + assert_eq!(status.error_code, "UPSTREAM_ERROR"); + assert_eq!( + status.message, + "UPSTREAM_ERROR: [Errno -3] Temporary failure in name resolution" + ); + assert_eq!(update.message, status.message); + Ok(()) + } + + #[test] + fn unreachable_discovery_backend_is_a_typed_unavailable_error() -> Result<(), AifarmDecodeError> + { + for code in [ + "UPSTREAM_ERROR", + "SERVICE_NOT_FOUND", + "SERVICE_DISABLED", + "ENDPOINT_NOT_FOUND", + ] { + let status = + discovery_status_from_envelope("job-1", &failed_discovery_envelope(code, "down"))?; + let error = discovery_result_from_status("job-1", &status, STATUS_FAILED) + .expect_err("a failed job is an error"); + assert!( + matches!( + error.downcast_ref::(), + Some(AifarmClientError::BackendUnavailable(message)) if message == &format!("{code}: down") + ), + "{code} must mean the backend is unavailable, got {error:?}" + ); + } + Ok(()) + } + + #[test] + fn discovery_timeout_and_unknown_failures_stay_upstream_errors() -> Result<(), AifarmDecodeError> + { + for (code, expected) in [ + ("UPSTREAM_TIMEOUT", "UPSTREAM_TIMEOUT: read timed out"), + ("INTERNAL_ERROR", "INTERNAL_ERROR: read timed out"), + ] { + let status = discovery_status_from_envelope( + "job-1", + &failed_discovery_envelope(code, "read timed out"), + )?; + let error = discovery_result_from_status("job-1", &status, STATUS_FAILED) + .expect_err("a failed job is an error"); + assert!( + matches!( + error.downcast_ref::(), + Some(AifarmClientError::Upstream(message)) if message == expected + ), + "{code} got {error:?}" + ); + } + + let bare = DiscoveryJobStatus { + status: STATUS_FAILED.to_owned(), + ..DiscoveryJobStatus::default() + }; + let error = discovery_result_from_status("job-1", &bare, STATUS_FAILED) + .expect_err("a failed job is an error"); + assert_eq!(error.to_string(), "dialog job failed"); + Ok(()) + } + #[test] fn message_body_uses_conversation_graph_metadata() { let body = format_message_body(&HistoryMessage { diff --git a/crates/openplotva-llm/src/retry.rs b/crates/openplotva-llm/src/retry.rs index 7bfeec7e..91265243 100644 --- a/crates/openplotva-llm/src/retry.rs +++ b/crates/openplotva-llm/src/retry.rs @@ -123,10 +123,13 @@ pub fn retryable_reason(err: &(dyn Error + 'static)) -> Option { if let Some(provider_error) = find_provider_error(err) { return Some(provider_error.reason()); } - // reqwest transport failures (connect, TLS, mid-body disconnects) stringify - // in ways the message rules can't enumerate; the typed variant is authoritative. - if let Some(crate::aifarm::AifarmClientError::Transport(_)) = - find_error::(err) + // reqwest transport failures (connect, TLS, mid-body disconnects) and Discovery's + // unreachable-backend codes stringify in ways the message rules can't enumerate; + // the typed variants are authoritative. + if let Some( + crate::aifarm::AifarmClientError::Transport(_) + | crate::aifarm::AifarmClientError::BackendUnavailable(_), + ) = find_error::(err) { return Some(FailureReason::ProviderUnavailable); } @@ -457,6 +460,20 @@ mod tests { ); } + #[test] + fn unreachable_discovery_backend_is_retryable_via_typed_source() { + // The DNS failure Discovery reported while ninfer-3090 was down matches no + // phrase rule; the typed variant alone must classify it. + let err = crate::aifarm::AifarmClientError::BackendUnavailable( + "UPSTREAM_ERROR: [Errno -3] Temporary failure in name resolution".to_owned(), + ); + + assert_eq!( + retryable_reason(&err), + Some(FailureReason::ProviderUnavailable) + ); + } + #[test] fn transport_body_decode_messages_are_retryable() { for message in [