diff --git a/crates/openplotva-app/src/embedder.rs b/crates/openplotva-app/src/embedder.rs index 8790e45..90a56d2 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 c9363c1..633103e 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 0eafaeb..862b0d0 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 ba1d70a..09f1a13 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 7bfeec7..9126524 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 [