fix(graph): typed ApiStatusError transient classification; warn on bare output_schema
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.
This commit is contained in:
2 files changed
+278
-7
No files matched your search
+86
-5
@@ -35,6 +35,8 @@ pub(crate) fn wall_clock(secs: u64) -> Option<Duration> {
|
||||
(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::<reqwest::Error>()
|
||||
@@ -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::<crate::client::ApiStatusError>())
|
||||
{
|
||||
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::<()>())
|
||||
|
||||
+192
-2
@@ -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<Stri
|
||||
}
|
||||
}
|
||||
|
||||
fn lifts_via_state_updates(updates: &Option<std::collections::HashMap<String, String>>) -> bool {
|
||||
updates.as_ref().is_some_and(|m| !m.is_empty())
|
||||
}
|
||||
|
||||
fn node_state_updates_keys(node: &Node) -> Option<HashSet<String>> {
|
||||
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
|
||||
);
|
||||
}
|
||||
}
|
||||
Reference in new issue
Block a user