fix(triage): defer malformed cloud replies (#2415)

src/openhuman/agent/triage/evaluator.rs

@@ -15,8 +15,10 @@
 //! ```
 //!
 //! Non-transient cloud failures (auth, malformed prompt, model not
-//! found, parse failure) bubble up immediately — there's no point
-//! retrying them and the local arm wouldn't help either.
+//! found) bubble up immediately — there's no point retrying them and
+//! the local arm wouldn't help either. Malformed classifier replies
+//! are treated like retryable cloud failures: retry once, then fall
+//! through to local / Deferred.
 //!
 //! ## Why `run_tool_call_loop` doesn't care about `tools_registry = []`
 //!
@@ -511,20 +513,24 @@ async fn try_arm(
             );
             // A parse failure means the model produced unusable
             // output. Retrying the same arm with the same prompt
-            // won't help, but on the *cloud* arm a parse failure is
-            // worth retrying once because the cloud model can be
-            // non-deterministic across calls. On the local arm we've
-            // already exhausted cloud and would just spin — treat it
-            // as fatal so the chain progresses to Deferred.
+            // won't usually help, but on the cloud arm one retry is
+            // cheap enough because hosted models can be
+            // non-deterministic across calls. If the cloud retry also
+            // returns malformed output, let the outer chain fall
+            // through to local/Deferred instead of surfacing Err to
+            // background callers like Composio trigger triage.
             return Err(match intended_path {
-                TriageResolutionPath::Cloud => ArmError::Retryable {
-                    retry_after_ms: None,
-                    source: anyhow!(
-                        "classifier reply did not parse: {}",
-                        format_parse_error(&parse_err)
-                    ),
-                },
-                _ => ArmError::Fatal(anyhow!(
+                TriageResolutionPath::Cloud | TriageResolutionPath::CloudAfterRetry => {
+                    ArmError::Retryable {
+                        retry_after_ms: None,
+                        source: anyhow!(
+                            "classifier reply did not parse on {} arm: {}",
+                            intended_path.as_str(),
+                            format_parse_error(&parse_err)
+                        ),
+                    }
+                }
+                TriageResolutionPath::LocalFallback => ArmError::Fatal(anyhow!(
                     "classifier reply did not parse on {} arm: {}",
                     intended_path.as_str(),
                     format_parse_error(&parse_err)

src/openhuman/agent/triage/evaluator_tests.rs

@@ -768,3 +768,89 @@ async fn no_local_arm_returns_deferred_after_cloud_exhaustion() {
         "1 cloud + 1 retry, no local"
     );
 }
+
+#[tokio::test]
+async fn double_cloud_parse_failure_falls_through_to_local_fallback() {
+    // Regression for #2322: two malformed cloud replies used to turn the
+    // second cloud parse error into ArmError::Fatal, bubbling out of
+    // run_triage as Err and making the Composio subscriber emit
+    // `[composio][triage] run_triage failed` at error level.
+    AgentDefinitionRegistry::init_global_builtins().expect("init_global_builtins");
+    let counter = StdArc::new(AtomicUsize::new(0));
+    let counter_for_stub = StdArc::clone(&counter);
+
+    let _guard = mock_agent_run_turn(move |req| {
+        let counter = StdArc::clone(&counter_for_stub);
+        async move {
+            let n = counter.fetch_add(1, Ordering::SeqCst);
+            if n < 2 {
+                assert_eq!(
+                    req.provider_name, "stub-cloud",
+                    "first two attempts should stay on the cloud arm"
+                );
+                Ok(AgentTurnResponse {
+                    text: "not json".to_string(),
+                })
+            } else {
+                assert_eq!(
+                    req.provider_name, "stub-local",
+                    "malformed cloud retry should fall through to local"
+                );
+                Ok(AgentTurnResponse {
+                    text: VALID_JSON_REPLY.to_string(),
+                })
+            }
+        }
+    })
+    .await;
+
+    let outcome = run_triage_with_arms_for_test(cloud_arm(), Some(local_arm()), &envelope())
+        .await
+        .expect("malformed cloud retry must fall through, not surface Err");
+
+    let run = outcome.into_decision().expect("decision");
+    assert_eq!(run.resolution_path, TriageResolutionPath::LocalFallback);
+    assert!(run.used_local);
+    assert_eq!(
+        counter.load(Ordering::SeqCst),
+        3,
+        "1 cloud + 1 cloud retry + 1 local"
+    );
+}
+
+#[tokio::test]
+async fn double_cloud_parse_failure_without_local_returns_deferred_not_err() {
+    AgentDefinitionRegistry::init_global_builtins().expect("init_global_builtins");
+    let counter = StdArc::new(AtomicUsize::new(0));
+    let counter_for_stub = StdArc::clone(&counter);
+
+    let _guard = mock_agent_run_turn(move |_req| {
+        let counter = StdArc::clone(&counter_for_stub);
+        async move {
+            counter.fetch_add(1, Ordering::SeqCst);
+            Ok(AgentTurnResponse {
+                text: "still not json".to_string(),
+            })
+        }
+    })
+    .await;
+
+    let outcome = run_triage_with_arms_for_test(cloud_arm(), None, &envelope())
+        .await
+        .expect("malformed cloud retry with no local must Defer, not Err");
+
+    match outcome {
+        TriageOutcome::Deferred { reason, .. } => {
+            assert!(
+                reason.contains("local arm unavailable"),
+                "reason should explain the missing local arm: {reason}"
+            );
+        }
+        TriageOutcome::Decision(_) => panic!("expected Deferred"),
+    }
+    assert_eq!(
+        counter.load(Ordering::SeqCst),
+        2,
+        "1 cloud + 1 cloud retry, no local"
+    );
+}