fix(flare-workflow): honor RetryPolicy for StepMode::Loop iterations

crates/flare-workflow/src/loops.rs

@@ -12,6 +12,7 @@ use tokio::time::timeout;
 
 use crate::definition::{StepDefinition, WorkflowDefinition};
 use crate::engine::WorkflowEngine;
+use crate::retry::{self, Backoff};
 use crate::store::StateStore;
 use crate::types::*;
 use crate::variables::capture_output;
@@ -81,7 +82,53 @@ impl<D: WorkflowData, S: StateStore<D> + 'static> WorkflowEngine<D, S> {
                 })
                 .await?;
 
-            let result = timeout(step_timeout, step.executor.execute(&mut context)).await;
+            // Retry the executor within this loop iteration per the step's
+            // `RetryPolicy` (falling back to the workflow's default, which
+            // is itself `max_attempts: 3` — see `RetryPolicy::default()`).
+            // Loop-mode steps used to ignore this entirely: `execute_loop`
+            // journaled and returned on the very first failure, so a
+            // step's configured retries never fired (item #486). Reuse the
+            // same in-memory `context` across attempts rather than
+            // reloading from `state_store` — a failed attempt's partial
+            // progress (e.g. `sdd_loop`'s role-agent reply already
+            // obtained before its judge call failed) isn't persisted
+            // either way, so keeping it in memory means a retry only
+            // redoes the part that actually failed.
+            let retry_policy = definition.get_retry_policy(step);
+            let max_retry_attempts = retry::effective_max_attempts(
+                retry_policy.max_attempts,
+                matches!(step.on_failure, FailureAction::RetryIndefinitely),
+            );
+            let mut backoff = Backoff::from_strategy(&retry_policy.backoff);
+            let mut retry_attempt = 1;
+            let result = loop {
+                let attempt_result =
+                    timeout(step_timeout, step.executor.execute(&mut context)).await;
+                let is_failure = matches!(
+                    attempt_result,
+                    Ok(Ok(StepResult::Failure)) | Ok(Err(_)) | Err(_)
+                );
+                if !is_failure {
+                    break attempt_result;
+                }
+                let retryable = match &attempt_result {
+                    Ok(Err(e)) => step.executor.is_retryable(e),
+                    Err(_) => true,
+                    _ => false,
+                };
+                if !retryable || retry_attempt >= max_retry_attempts {
+                    break attempt_result;
+                }
+                let delay = backoff
+                    .next(self.jitter)
+                    .unwrap_or_else(|| std::time::Duration::from_secs(1));
+                tracing::warn!(
+                    run_id = %run_id, step = %step.id, iter, retry_attempt, ?delay,
+                    "Loop iteration step failed, retrying"
+                );
+                tokio::time::sleep(delay).await;
+                retry_attempt += 1;
+            };
             let duration_ms = step_start.elapsed().as_millis() as u64;
 
             match result {

crates/flare-workflow/tests/semantics_test.rs

@@ -157,6 +157,99 @@ async fn loop_respects_max_iterations() {
     assert_eq!(state.context.data.calls.len(), 3);
 }
 
+#[tokio::test]
+async fn loop_iteration_retries_transient_failure_then_succeeds() {
+    // item #486: `execute_loop` used to ignore the step's `RetryPolicy`
+    // entirely, hard-failing the whole loop on the very first error.
+    let calls = Arc::new(AtomicU32::new(0));
+    let cc = Arc::clone(&calls);
+    let wf = WorkflowDefinition::new("wf", "wf").add_step(
+        StepDefinition::new(
+            "flaky",
+            "flaky",
+            Arc::new(FunctionStep::new(move |ctx: &mut WorkflowContext<Ctx>| {
+                let n = cc.fetch_add(1, Ordering::SeqCst);
+                let fail = n < 2;
+                if !fail {
+                    ctx.data.calls.push(format!("attempt{n}"));
+                    ctx.output = "Result: DONE".to_string();
+                }
+                Box::pin(async move {
+                    if fail {
+                        Err(WorkflowError::StepFailed {
+                            step_id: StepId::new("flaky"),
+                            message: "transient".into(),
+                        })
+                    } else {
+                        Ok(StepResult::Success)
+                    }
+                })
+            })),
+        )
+        .with_mode(StepMode::Loop {
+            max_iterations: 5,
+            until: "DONE".into(),
+        })
+        .with_retry(RetryPolicy {
+            max_attempts: 3,
+            backoff: BackoffStrategy::Fixed(Duration::from_millis(1)),
+        }),
+    );
+    let state = run_and_wait(&engine(), wf).await;
+    assert_eq!(state.status, WorkflowStatus::Completed);
+    // 2 failures + 1 success, all retried within the SAME loop iteration
+    // (the loop's iteration counter never advances on a retried attempt).
+    assert_eq!(calls.load(Ordering::SeqCst), 3);
+    assert_eq!(state.context.data.calls, vec!["attempt2"]);
+    assert!(state.input.contains("DONE"));
+}
+
+#[tokio::test]
+async fn loop_iteration_fails_after_exhausting_retries() {
+    let calls = Arc::new(AtomicU32::new(0));
+    let cc = Arc::clone(&calls);
+    let wf = WorkflowDefinition::new("wf", "wf").add_step(
+        StepDefinition::new(
+            "always-fails",
+            "always-fails",
+            Arc::new(FunctionStep::new(move |_ctx: &mut WorkflowContext<Ctx>| {
+                cc.fetch_add(1, Ordering::SeqCst);
+                Box::pin(async move {
+                    Err(WorkflowError::StepFailed {
+                        step_id: StepId::new("always-fails"),
+                        message: "always broken".into(),
+                    })
+                })
+            })),
+        )
+        .with_mode(StepMode::Loop {
+            max_iterations: 5,
+            until: "DONE".into(),
+        })
+        .with_retry(RetryPolicy {
+            max_attempts: 2,
+            backoff: BackoffStrategy::Fixed(Duration::from_millis(1)),
+        }),
+    );
+    let e = engine();
+    e.register_workflow(wf).unwrap();
+    let run = e
+        .start_workflow(WorkflowId::new("wf"), Ctx { calls: vec![] }, "seed".into())
+        .await
+        .unwrap();
+    let out = e
+        .wait_for_completion(run, "wf", Duration::from_secs(10))
+        .await;
+    assert!(out.is_err());
+
+    // Exactly `max_attempts` executor calls -- retries exhausted within
+    // the first iteration, no further iterations attempted, and it still
+    // terminates the workflow as a failure (preserving prior behavior).
+    assert_eq!(calls.load(Ordering::SeqCst), 2);
+    let state = e.get_status(run).await.unwrap();
+    assert_eq!(state.status, WorkflowStatus::Failed);
+}
+
 #[tokio::test]
 async fn fan_out_collect_joins_outputs() {
     let wf = WorkflowDefinition::new("wf", "wf")