From 8ea18bdd0234eb9047ac57013027e63adddebc73 Mon Sep 17 00:00:00 2001 From: Shivakumar Date: Sun, 16 Aug 2026 11:37:14 +0530 Subject: [PATCH 1/2] fix(flare-workflow): honor RetryPolicy for StepMode::Loop iterations execute_loop dispatched from StepMode::Loop never consulted the step's RetryPolicy, hard-failing the whole loop on the first error even when a policy (including the crate default) configured multiple attempts. sdd_loop's configured 3-attempt retry for malformed judge replies was silently dead code as a result. Retry the current iteration in place (not advancing iter) up to the policy's max_attempts with backoff, reusing the same in-memory context across attempts so a partially-completed iteration isn't redone. Agentflare-Agent: claude-code_2-1-233_agent Agentflare-Branch: task/486-flare-workflow-execute-loop-silently-ign Agentflare-Item: 486 --- crates/flare-workflow/src/loops.rs | 48 +++++++++- crates/flare-workflow/tests/semantics_test.rs | 93 +++++++++++++++++++ 2 files changed, 140 insertions(+), 1 deletion(-) diff --git a/crates/flare-workflow/src/loops.rs b/crates/flare-workflow/src/loops.rs index fb08899a..4a52ef36 100644 --- a/crates/flare-workflow/src/loops.rs +++ b/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,52 @@ impl + 'static> WorkflowEngine { }) .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 { diff --git a/crates/flare-workflow/tests/semantics_test.rs b/crates/flare-workflow/tests/semantics_test.rs index e1f37e21..1391d49c 100644 --- a/crates/flare-workflow/tests/semantics_test.rs +++ b/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| { + 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| { + 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") From 78b45073b3593679ed225ba20839268017a10d0e Mon Sep 17 00:00:00 2001 From: Shivakumar Date: Sun, 16 Aug 2026 13:23:24 +0530 Subject: [PATCH 2/2] style: cargo fmt loops.rs Agentflare-Agent: claude-code_2-1-233_agent Agentflare-Branch: task/486-flare-workflow-execute-loop-silently-ign Agentflare-Item: 486 --- crates/flare-workflow/src/loops.rs | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/crates/flare-workflow/src/loops.rs b/crates/flare-workflow/src/loops.rs index 4a52ef36..1b55bfa2 100644 --- a/crates/flare-workflow/src/loops.rs +++ b/crates/flare-workflow/src/loops.rs @@ -102,7 +102,8 @@ impl + 'static> WorkflowEngine { 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 attempt_result = + timeout(step_timeout, step.executor.execute(&mut context)).await; let is_failure = matches!( attempt_result, Ok(Ok(StepResult::Failure)) | Ok(Err(_)) | Err(_)