diff --git a/src/client/common.rs b/src/client/common.rs index e652499..ac51ee6 100644 --- a/src/client/common.rs +++ b/src/client/common.rs @@ -597,7 +597,7 @@ pub async fn call_chat_completions_streaming( } /// Streaming transport for graph `llm` nodes: accumulates the reply and the -/// transport itself renders nothing — no spinner, delta, or reasoning +/// transport itself renders nothing; no spinner, delta, or reasoning /// (tool-call rendering inside `eval_tool_calls` is unchanged from the /// non-streaming path). Because chunks arrive as they are generated, the /// reqwest `read_timeout` bounds only stalls between chunks; the caller's @@ -605,7 +605,7 @@ pub async fn call_chat_completions_streaming( /// transport is kept deliberately: the handler's call-loop detection is off /// (that path never had it), and a model the catalog marks `no_stream` /// takes the non-streaming request, raced against the abort. An abort -/// observed after the provider call fails outright — partial output is +/// observed after the provider call fails outright. Partial output is /// never returned as success. pub async fn call_chat_completions_streaming_quiet( input: &Input, @@ -900,8 +900,10 @@ mod tests { use super::*; use super::super::access_token::{is_rejected, set_access_token}; + use crate::config::{AppState, WorkingMode}; use std::sync::Arc; use std::sync::atomic::{AtomicUsize, Ordering}; + use tokio::time::Instant; fn catch_error_message(data: &Value, status: u16) -> String { catch_error(data, status).unwrap_err().to_string() @@ -1189,10 +1191,7 @@ mod tests { } fn quiet_ctx() -> RequestContext { - RequestContext::new( - Arc::new(crate::config::AppState::test_default()), - crate::config::WorkingMode::Cmd, - ) + RequestContext::new(Arc::new(AppState::test_default()), WorkingMode::Cmd) } /// 600s of virtual generation completes and yields the same shape the @@ -1207,7 +1206,7 @@ mod tests { let mut ctx = quiet_ctx(); let input = Input::from_str(&ctx, "hi", None).unwrap(); let client = paced_client(false); - let started = tokio::time::Instant::now(); + let started = Instant::now(); let (text, results) = call_chat_completions_streaming_quiet(&input, &client, &mut ctx, create_abort_signal()) @@ -1236,7 +1235,7 @@ mod tests { tokio::time::sleep(PACED_GAP * PACED_DELTAS as u32 + Duration::from_secs(30)).await; trigger.set_ctrlc(); }); - let started = tokio::time::Instant::now(); + let started = Instant::now(); let err = call_chat_completions_streaming_quiet(&input, &client, &mut ctx, abort) .await @@ -1312,7 +1311,7 @@ mod tests { tokio::time::sleep(Duration::from_secs(5)).await; trigger.set_ctrlc(); }); - let started = tokio::time::Instant::now(); + let started = Instant::now(); let err = call_chat_completions_streaming_quiet(&input, &client, &mut ctx, abort) .await @@ -1326,14 +1325,14 @@ mod tests { /// transport still streams with it off. #[tokio::test(start_paused = true)] async fn quiet_transport_ignores_the_user_stream_toggle() { - let app = crate::config::AppState { + let app = AppState { config: Arc::new(AppConfig { stream: false, ..AppConfig::default() }), - ..crate::config::AppState::test_default() + ..AppState::test_default() }; - let mut ctx = RequestContext::new(Arc::new(app), crate::config::WorkingMode::Cmd); + let mut ctx = RequestContext::new(Arc::new(app), WorkingMode::Cmd); let input = Input::from_str(&ctx, "hi", None).unwrap(); assert!(!input.stream(), "fixture must have streaming disabled"); let client = paced_client(false); diff --git a/src/graph/agent.rs b/src/graph/agent.rs index 201011f..86ded61 100644 --- a/src/graph/agent.rs +++ b/src/graph/agent.rs @@ -127,7 +127,7 @@ async fn run( } /// Drives the retry loop, then extraction, then state updates. Extraction -/// stays outside the retry loop: by the time it runs the agent itself +/// stays outside the retry loop: by the time it runs, the agent itself /// succeeded, so an extraction failure is never retried. async fn attempt_and_extract( node_id: &str, @@ -249,6 +249,7 @@ async fn retry_transient( parent_ctx.peer_registry = Some(Arc::clone(registry)); } } + match run_attempt(parent_ctx).await { Ok(out) => return Ok(out), Err(e) if is_transient_error(&e) && attempt < node.max_attempts => { @@ -260,6 +261,7 @@ async fn retry_transient( Err(e) => return Err(e), } } + Err(last_err.unwrap_or_else(|| anyhow!("agent node exhausted retries"))) } diff --git a/src/graph/mod.rs b/src/graph/mod.rs index 9e23879..8e17c6d 100644 --- a/src/graph/mod.rs +++ b/src/graph/mod.rs @@ -16,6 +16,7 @@ pub mod types; pub mod user_interaction; pub mod validator; +use crate::client::ApiStatusError; use anyhow::Error; pub use dispatch::{ active_agent_graph_name, run_active_agent_graph, run_active_agent_graph_with_inputs, @@ -45,10 +46,7 @@ pub(crate) fn is_transient_error(err: &Error) -> bool { }) { return true; } - if let Some(api) = err - .chain() - .find_map(|c| c.downcast_ref::()) - { + if let Some(api) = err.chain().find_map(|c| c.downcast_ref::()) { return is_transient_status(api.status); } let s = format!("{err:#}").to_lowercase(); @@ -67,8 +65,8 @@ pub(crate) fn is_transient_error(err: &Error) -> bool { /// Transient HTTP statuses worth a from-scratch retry: 429 (rate limit), /// 500/502/503/504 (upstream hiccups), 529 (Anthropic overloaded). -/// Deterministic statuses — 4xx other than 429 (including 408) and -/// 501/505-class capability errors — are excluded. +/// Deterministic statuses (4xx other than 429 (including 408) and +/// 501/505-class capability errors) are excluded. fn is_transient_status(status: u16) -> bool { matches!(status, 429 | 500 | 502 | 503 | 504 | 529) } @@ -89,6 +87,7 @@ pub(in crate::graph) fn type_name(value: &Value) -> &'static str { #[cfg(test)] mod tests { use super::*; + use crate::client; use anyhow::anyhow; #[test] @@ -151,7 +150,7 @@ mod tests { "message": "Too many requests, please slow down" } }); - let err = crate::client::catch_error(&data, 429) + let err = client::catch_error(&data, 429) .unwrap_err() .context("Failed to call chat-completions api") .context("Agent 'domain-reviewer' failed"); @@ -166,7 +165,7 @@ mod tests { #[test] fn is_transient_error_typed_status_overrides_message_text() { - let err = crate::client::catch_error( + let err = client::catch_error( &serde_json::json!({ "message": "rate limit reached — connection reset" }), 400, ) @@ -176,7 +175,7 @@ mod tests { "typed 400 must not be retried even when the body matches string anchors" ); - let err = crate::client::catch_error(&serde_json::json!({ "message": "x" }), 503) + let err = client::catch_error(&serde_json::json!({ "message": "x" }), 503) .unwrap_err() .context("Agent 'x' failed"); assert!(is_transient_error(&err)); @@ -186,14 +185,14 @@ mod tests { fn is_transient_error_typed_status_table() { let data = serde_json::json!({ "message": "x" }); for status in [429, 500, 502, 503, 504, 529] { - let err = crate::client::catch_error(&data, status).unwrap_err(); + let err = client::catch_error(&data, status).unwrap_err(); assert!( is_transient_error(&err), "status {status} should be transient" ); } for status in [400, 401, 403, 404, 408, 422, 501, 505] { - let err = crate::client::catch_error(&data, status).unwrap_err(); + let err = client::catch_error(&data, status).unwrap_err(); assert!( !is_transient_error(&err), "status {status} should not be transient" @@ -261,7 +260,7 @@ mod tests { .read_timeout(Duration::from_millis(50)) .build() .unwrap(); - let err = crate::client::sse_stream(client.get(format!("http://{addr}/")), |_| Ok(false)) + let err = client::sse_stream(client.get(format!("http://{addr}/")), |_| Ok(false)) .await .expect_err("a body that never arrives must trip the read timeout"); server.abort(); @@ -308,7 +307,7 @@ mod tests { .build() .unwrap(); let mut received = 0usize; - crate::client::sse_stream(client.get(format!("http://{addr}/")), |message| { + client::sse_stream(client.get(format!("http://{addr}/")), |message| { if message.data == "[DONE]" { return Ok(true); } diff --git a/src/graph/validator.rs b/src/graph/validator.rs index 0dc456d..69aa6d4 100644 --- a/src/graph/validator.rs +++ b/src/graph/validator.rs @@ -5,7 +5,7 @@ use crate::config; use crate::config::{Agent, AppConfig, paths}; use crate::rag::{GraphRagConfig, RagData}; use anyhow::{Result, bail}; -use std::collections::{BTreeMap, HashSet, VecDeque}; +use std::collections::{BTreeMap, HashMap, HashSet, VecDeque}; use std::path::PathBuf; use std::sync::Arc; use tokio::sync::Semaphore; @@ -451,7 +451,7 @@ impl GraphValidator { format!( "{kind} node's `output_schema` declares no (non-empty) object `properties`; \ the engine auto-merges ONLY keys declared under `properties` (a bare \ - object schema merges nothing) — declare the keys under \ + object schema merges nothing). Declare the keys under \ `properties` or lift them via `state_updates`" ), )); @@ -1297,7 +1297,7 @@ fn primary_templated_fields(node: &Node) -> Vec { } } -fn node_state_updates_map(node: &Node) -> Option<&std::collections::HashMap> { +fn node_state_updates_map(node: &Node) -> Option<&HashMap> { match &node.node_type { NodeType::Llm(n) => n.state_updates.as_ref(), NodeType::Agent(n) => n.state_updates.as_ref(), @@ -1310,7 +1310,7 @@ fn node_state_updates_map(node: &Node) -> Option<&std::collections::HashMap>) -> bool { +fn lifts_via_state_updates(updates: &Option>) -> bool { updates.as_ref().is_some_and(|m| !m.is_empty()) }