From ea76331c49c65da33aca75aac3f3107bd0584077 Mon Sep 17 00:00:00 2001 From: Alex Clarke Date: Sun, 13 Sep 2026 19:37:28 -0600 Subject: [PATCH] fix(graph): typed ApiStatusError transient classification; warn on bare output_schema MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit is_transient_error: a typed crate::client::ApiStatusError anywhere in the chain is now authoritative (positive and negative) — catch_error renders 429s as "… (type: rate_limit_error)" / "… (status: 429)", which none of the old string anchors matched, so provider rate limits were never retried, while a 400 whose body mentioned "rate limit" was. Transient statuses: 429, 500, 502, 503, 504, 529; message text only classifies untyped errors, and those anchors are now case-insensitive. validator: warn for an llm/agent node whose output_schema lacks a non-empty object `properties` map (auto-merge writes nothing since 402d25a) unless the node lifts keys via a non-empty state_updates; the same absent-or-empty predicate now also gates validate_fallback_capture. Bundled and graph.example.yaml warning baselines are unchanged. Tests build the 429 fixture through real catch_error output and a .context() chain, cover the status table, typed-overrides-text, and all validator branches. --- src/graph/mod.rs | 91 +++++++++++++++++-- src/graph/validator.rs | 194 ++++++++++++++++++++++++++++++++++++++++- 2 files changed, 278 insertions(+), 7 deletions(-) diff --git a/src/graph/mod.rs b/src/graph/mod.rs index 4b0dafc..9e23879 100644 --- a/src/graph/mod.rs +++ b/src/graph/mod.rs @@ -35,6 +35,8 @@ pub(crate) fn wall_clock(secs: u64) -> Option { (secs != 0).then(|| Duration::from_secs(secs)) } +/// An HTTP status in the chain is authoritative — message text never +/// overrides it. String anchors are only consulted for untyped errors. pub(crate) fn is_transient_error(err: &Error) -> bool { if err.chain().any(|c| { c.downcast_ref::() @@ -43,20 +45,34 @@ pub(crate) fn is_transient_error(err: &Error) -> bool { }) { return true; } - let s = format!("{err:#}"); + if let Some(api) = err + .chain() + .find_map(|c| c.downcast_ref::()) + { + return is_transient_status(api.status); + } + let s = format!("{err:#}").to_lowercase(); s.contains("timed out") || s.contains("rate limit") - || s.contains("HTTP 429") + || s.contains("http 429") || s.contains("status 429") - || s.contains("429 Too Many") - || s.contains("Connection reset") - || s.contains("Connection refused") + || s.contains("429 too many") + || s.contains("connection reset") + || s.contains("connection refused") || s.contains("produced no output") || s.contains("broken pipe") || s.contains("connection error") || s.contains("error sending request") } +/// 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. +fn is_transient_status(status: u16) -> bool { + matches!(status, 429 | 500 | 502 | 503 | 504 | 529) +} + pub const MAX_STATE_SIZE_BYTES: usize = 32 * 1024; pub(in crate::graph) fn type_name(value: &Value) -> &'static str { @@ -127,6 +143,71 @@ mod tests { ))); } + #[test] + fn is_transient_error_sees_typed_api_status_429_through_context_chain() { + let data = serde_json::json!({ + "error": { + "type": "rate_limit_error", + "message": "Too many requests, please slow down" + } + }); + let err = crate::client::catch_error(&data, 429) + .unwrap_err() + .context("Failed to call chat-completions api") + .context("Agent 'domain-reviewer' failed"); + + let rendered = format!("{err:#}"); + assert!( + !is_transient_error(&anyhow!("{rendered}")), + "the rendered text alone must not be transient — proves the typed arm is what catches it" + ); + assert!(is_transient_error(&err)); + } + + #[test] + fn is_transient_error_typed_status_overrides_message_text() { + let err = crate::client::catch_error( + &serde_json::json!({ "message": "rate limit reached — connection reset" }), + 400, + ) + .unwrap_err(); + assert!( + !is_transient_error(&err), + "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) + .unwrap_err() + .context("Agent 'x' failed"); + assert!(is_transient_error(&err)); + } + + #[test] + 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(); + 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(); + assert!( + !is_transient_error(&err), + "status {status} should not be transient" + ); + } + } + + #[test] + fn is_transient_error_string_anchors_are_case_insensitive() { + assert!(is_transient_error(&anyhow!("Rate Limit reached"))); + assert!(is_transient_error(&anyhow!("Request Timed Out"))); + assert!(is_transient_error(&anyhow!("connection reset by peer"))); + } + #[tokio::test(start_paused = true)] async fn is_transient_error_recognizes_typed_elapsed_without_matching_text() { let elapsed = tokio::time::timeout(Duration::from_secs(1), std::future::pending::<()>()) diff --git a/src/graph/validator.rs b/src/graph/validator.rs index 3c5d6d0..0dc456d 100644 --- a/src/graph/validator.rs +++ b/src/graph/validator.rs @@ -178,6 +178,7 @@ impl GraphValidator { self.validate_llm_skills(graph, &mut result); self.validate_max_attempts(graph, &mut result); self.validate_fallback_capture(graph, &mut result); + self.validate_output_schema_properties(graph, &mut result); self.validate_max_concurrency(graph, &mut result); self.validate_max_concurrency_template(graph, &mut result); self.validate_orchestration_limits(graph, &mut result); @@ -399,14 +400,14 @@ impl GraphValidator { NodeType::Llm(l) if l.fallback.is_some() && l.output_schema.is_some() - && l.state_updates.is_none() => + && !lifts_via_state_updates(&l.state_updates) => { "llm" } NodeType::Agent(a) if a.fallback.is_some() && a.output_schema.is_some() - && a.state_updates.is_none() => + && !lifts_via_state_updates(&a.state_updates) => { "agent" } @@ -424,6 +425,39 @@ impl GraphValidator { } } + fn validate_output_schema_properties(&self, graph: &Graph, result: &mut ValidationResult) { + for (node_id, node) in &graph.nodes { + let (kind, schema) = match &node.node_type { + NodeType::Llm(l) if !lifts_via_state_updates(&l.state_updates) => { + ("llm", l.output_schema.as_ref()) + } + NodeType::Agent(a) if !lifts_via_state_updates(&a.state_updates) => { + ("agent", a.output_schema.as_ref()) + } + _ => continue, + }; + let Some(schema) = schema else { + continue; + }; + if schema + .get("properties") + .and_then(serde_json::Value::as_object) + .is_some_and(|m| !m.is_empty()) + { + continue; + } + result.warning(ValidationError::with_node( + node_id, + 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 \ + `properties` or lift them via `state_updates`" + ), + )); + } + } + fn validate_node_references(&self, graph: &Graph, result: &mut ValidationResult) { for (node_id, node) in &graph.nodes { for (target, label) in declared_targets(node) { @@ -1276,6 +1310,10 @@ fn node_state_updates_map(node: &Node) -> Option<&std::collections::HashMap>) -> bool { + updates.as_ref().is_some_and(|m| !m.is_empty()) +} + fn node_state_updates_keys(node: &Node) -> Option> { let updates = match &node.node_type { NodeType::Agent(n) => n.state_updates.as_ref(), @@ -4647,6 +4685,36 @@ mod tests { ); } + #[test] + fn fallback_with_output_schema_and_empty_state_updates_warns() { + let mut node = llm_with_output_schema("l", &["summary"], Some("end")); + if let NodeType::Llm(ref mut n) = node.node_type { + n.fallback = Some("fb".into()); + n.state_updates = Some(HashMap::new()); + } + let graph = graph_with( + vec![ + ("l", node), + ("fb", end_node("fb")), + ("end", end_node("end")), + ], + "l", + ); + + let result = validator().validate(&graph); + + assert!( + result + .warnings + .iter() + .any(|e| e.node_id.as_deref() == Some("l") + && e.message + .contains("declares `fallback` with `output_schema`")), + "expected fallback capture warning: {:?}", + result.warnings + ); + } + #[test] fn output_schema_without_fallback_is_not_flagged() { let graph = graph_with( @@ -4667,4 +4735,126 @@ mod tests { result.warnings ); } + + fn bare_object_schema_warnings(result: &ValidationResult) -> Vec<&ValidationError> { + result + .warnings + .iter() + .filter(|e| { + e.message + .contains("declares no (non-empty) object `properties`") + }) + .collect() + } + + #[test] + fn llm_output_schema_without_properties_warns() { + let mut node = llm_node("l", None, Some("end")); + if let NodeType::Llm(ref mut n) = node.node_type { + n.output_schema = Some(serde_json::json!({ "type": "object" })); + } + let graph = graph_with(vec![("l", node), ("end", end_node("end"))], "l"); + + let result = validator().validate(&graph); + + let warnings = bare_object_schema_warnings(&result); + assert_eq!(warnings.len(), 1, "{:?}", result.warnings); + assert_eq!(warnings[0].node_id.as_deref(), Some("l")); + assert!(warnings[0].message.starts_with("llm node's")); + } + + #[test] + fn agent_output_schema_without_properties_warns() { + let mut node = agent_node("a", "helper", Some("end")); + if let NodeType::Agent(ref mut n) = node.node_type { + n.output_schema = Some(serde_json::json!({ "type": "object" })); + } + let graph = graph_with(vec![("a", node), ("end", end_node("end"))], "a"); + + let result = validator().validate(&graph); + + let warnings = bare_object_schema_warnings(&result); + assert_eq!(warnings.len(), 1, "{:?}", result.warnings); + assert_eq!(warnings[0].node_id.as_deref(), Some("a")); + assert!(warnings[0].message.starts_with("agent node's")); + } + + #[test] + fn output_schema_with_empty_or_non_object_properties_warns() { + for properties in [serde_json::json!({}), serde_json::json!([])] { + let mut node = llm_node("l", None, Some("end")); + if let NodeType::Llm(ref mut n) = node.node_type { + n.output_schema = Some(serde_json::json!({ + "type": "object", + "properties": properties, + })); + } + let graph = graph_with(vec![("l", node), ("end", end_node("end"))], "l"); + + let result = validator().validate(&graph); + + let warnings = bare_object_schema_warnings(&result); + assert_eq!( + warnings.len(), + 1, + "properties={properties}: {:?}", + result.warnings + ); + assert_eq!(warnings[0].node_id.as_deref(), Some("l")); + } + } + + #[test] + fn bare_output_schema_with_state_updates_is_not_flagged() { + let mut node = llm_node("l", None, Some("end")); + if let NodeType::Llm(ref mut n) = node.node_type { + n.output_schema = Some(serde_json::json!({ "type": "object" })); + n.state_updates = Some(HashMap::from([("k".into(), "{{output}}".into())])); + } + let graph = graph_with(vec![("l", node), ("end", end_node("end"))], "l"); + + let result = validator().validate(&graph); + + assert!( + bare_object_schema_warnings(&result).is_empty(), + "{:?}", + result.warnings + ); + } + + #[test] + fn bare_output_schema_with_empty_state_updates_still_warns() { + let mut node = llm_node("l", None, Some("end")); + if let NodeType::Llm(ref mut n) = node.node_type { + n.output_schema = Some(serde_json::json!({ "type": "object" })); + n.state_updates = Some(HashMap::new()); + } + let graph = graph_with(vec![("l", node), ("end", end_node("end"))], "l"); + + let result = validator().validate(&graph); + + let warnings = bare_object_schema_warnings(&result); + assert_eq!(warnings.len(), 1, "{:?}", result.warnings); + assert_eq!(warnings[0].node_id.as_deref(), Some("l")); + } + + #[test] + fn output_schema_with_properties_or_absent_is_not_flagged() { + let graph = graph_with( + vec![ + ("l", llm_with_output_schema("l", &["k"], Some("plain"))), + ("plain", llm_node("plain", None, Some("end"))), + ("end", end_node("end")), + ], + "l", + ); + + let result = validator().validate(&graph); + + assert!( + bare_object_schema_warnings(&result).is_empty(), + "{:?}", + result.warnings + ); + } }