Checkpoint Resume Not Re-Executing Failed Executors

[iforkit]

Summary
When resuming a workflow from a checkpoint after a failure, the failed executor doesn't re-execute. The workflow appears to skip directly to completion, treating the previously-failed executor as if it had succeeded.

Environment

• Framework: Multi-Agent Framework (MAF)
• Checkpoint Storage: InMemoryCheckpointStorage (custom TrackingCheckpointStorage wrapper)
• Workflow Type: Sequential pipeline with 5 executors

The Problem
What I'm Trying to Do
I have a sequential order processing workflow with these steps:

1. Tool-1 - Fetch pricing data
2. Tool-2 - Execute validation rules
3. Tool-3 - Convert rules to natural language
4. Tool-5 - Save results to database
5. workflow_complete - Terminal node

When an executor fails (e.g., step 3 throws an exception), I want to:

1. Capture the last successful checkpoint (after step 2 completes)
2. On retry, resume from that checkpoint
3. Have the workflow re-execute step 3 and continue from there

What Actually Happens

When I resume from a checkpoint:

• The workflow stream completes immediately
• No executors re-execute
• No events are emitted (no ExecutorStartedEvent, no state updates)
• The workflow acts as if everything already succeeded

Expected Behavior

When resuming from a checkpoint, I expect:

• The workflow to identify which executor failed
• Re-execute that executor and all subsequent executors
• Emit normal workflow events (ExecutorStartedEvent, MessageSentEvent, etc.)
• Continue processing until completion or another failure

Code Example

Checkpoint Capture

async for event in stream:
    # Update state from workflow events
    if getattr(event, "data", None) is not None and isinstance(event.data, WorkflowState):
        state = event.data

    # Update safe checkpoint after superstep completion
    if isinstance(event, SuperStepCompletedEvent) and storage.last_checkpoint_id:
        safe_checkpoint_id = storage.last_checkpoint_id
        logger.info(f"[SAFE_CHECKPOINT] {safe_checkpoint_id}")

    # Detect failures
    if isinstance(event, (ExecutorFailedEvent, WorkflowFailedEvent)):
        last_failure = classify_event_failure(event)
        logger.warning(f"[FAILURE_DETECTED] {last_failure.reason}")

Resume Attempt

# On retry (attempt 2, 3, etc.)
if safe_checkpoint_id:
    logger.info(f"Resuming from checkpoint {safe_checkpoint_id}")
    stream = workflow.run_stream(
        checkpoint_id=safe_checkpoint_id,
        checkpoint_storage=storage
    )

    async for event in stream:
        # Stream completes immediately - no events emitted!
        logger.info(f"Event: {type(event).__name__}")

Workflow Structure

@executor(id="fetch_pricing")
async def get_fetch_pricing_executor(state: WorkflowState, ctx: WorkflowContext) -> None:
    fetch_pricing = await tools.get_all_fetch_pricing(state.order)
    state.fetch_pricing = fetch_pricing
    await ctx.send_message(state)

@executor(id="run_biz_rules")
async def run_biz_rules_executor(state: WorkflowState, ctx: WorkflowContext) -> None:
    rule_trace = await tools.run_biz_rules(state.order, state.average_prices)
    state.rule_trace = rule_trace
    await ctx.send_message(state)

# ... more executors ...

builder = WorkflowBuilder()
builder.set_start_executor(get_fetch_pricing_executor)
builder.add_edge(get_fetch_pricing_executor, run_biz_rules_executor)
builder.add_edge(run_biz_rules_executor, trace_executor)
builder.add_edge(trace_executor, save_db_executor)
builder.add_edge(save_db_executor, workflow_complete_executor)

return builder.with_checkpointing(checkpoint_storage).build()

Observed Logs

First Attempt (Initial Run)

[START] attempt=1/3 - initial workflow run
[SAFE_CHECKPOINT:init] checkpoint_abc123
[ExecutorStartedEvent] executor=get_fetch_pricing
[MessageSentEvent] executor=get_fetch_pricing
[SuperStepCompletedEvent]
[SAFE_CHECKPOINT] checkpoint_def456
[ExecutorStartedEvent] executor=run_biz_rules
[MessageSentEvent] executor=run_biz_rules
[SuperStepCompletedEvent]
[SAFE_CHECKPOINT] checkpoint_ghi789
[ExecutorStartedEvent] executor=rule_trace
[ExecutorFailedEvent] executor=rule_trace error="API timeout"
[FAILURE_DETECTED] API timeout

Second Attempt (Resume)

[RESUME] attempt=2/3 - resuming from checkpoint ghi789
[WorkflowCompletedEvent]

That's it. No executor events, no re-execution.

What I've Tried

1. Using run_from_checkpoint() instead of run_stream()

   • Same behavior - immediate completion with no executor re-execution

2. Capturing different checkpoints

   • Tried initial checkpoint, tried checkpoint before failure
   • No difference in behavior

3. Creating new workflow instance per retry

   • Thought maybe workflow state was cached
   • Same issue persists

4. Adding a terminal executor

   • Ensured failed executor is non-terminal (has outgoing edges)
   • Still doesn't re-execute

5. Verifying checkpoint contains state

   • Confirmed checkpoint has correct state data
   • State restoration works, but executors don't run

Questions

1. Is run_stream(checkpoint_id=X, checkpoint_storage=storage) the correct API for resuming?

   • Should I use a different method?
   • Is there a separate "resume" vs "restore" concept?

2. Do I need to configure executors differently for checkpoint resume?

   • Is there a flag or decorator parameter I'm missing?
   • Do executors need to be idempotent in a specific way?

3. Does checkpoint resume require workflow state to be structured differently?

   • Are there requirements about what goes in state vs context?
   • Should I be using workflow context differently?

4. Is there a way to debug which checkpoint version is loaded?

   • How can I verify the checkpoint is actually being used?
   • Is there logging I can enable to see checkpoint restoration?

What Would Help

1. A working example of checkpoint resume with executor re-execution
2. Documentation explaining the expected behavior of checkpoint resume
3. Guidance on proper checkpoint capture timing (before vs after executor completion)
4. Any known issues or common mistakes with checkpoint resume

---

Additional Context

I'm implementing automatic retry with checkpoint resume to avoid redundant work. The workflow makes external API calls that are expensive/slow, so I want to:

• On transient failures (timeouts, 5xx errors): Resume from last successful step
• On permanent failures (auth errors, validation): Fail immediately without retry

The checkpoint system seems to save state correctly, but the resume mechanism doesn't trigger executor re-execution, making checkpoints ineffective for retry scenarios.

Any guidance would be greatly appreciated!

[moonbox3]

I've created an issue from this so it's easier to track: #3758. Please follow along there. Closing this discussion.