Manual Recovery
Escalate a Step to a person after its automatic retries are exhausted.
ManualRecoveryFlow starts DoWorkStep. Its failure policy makes three exponential-backoff retries, then proceeds to ManualStep. ManualStep waits for either the Retry Channel or the Skip Channel. Retry returns to DoWorkStep after the operator resolves the problem; skip fails the Flow with an explicit reason.
Core implementation
DoWorkStep owns the retry policy and its exhausted-failure movement. ManualStep owns the durable wait for the operator's Channel message and turns that message into the next decision.
class DoWorkStep(Step[bool]):
def get_step_options(self) -> StepOptions:
return StepOptions(
execute_retry=RetryPolicy(
initial_interval=timedelta(seconds=1),
backoff_coefficient=2.0,
maximum_interval=timedelta(seconds=4),
maximum_attempts=4,
)
).on_execute_failure_proceed_to(ManualStep)
def execute(self, context: Context, should_fail: bool) -> StepDecision:
if should_fail:
raise RuntimeError("work failed")
return graceful_complete("work completed")
class ManualStep(Step[bool]):
def wait_for(self, context: Context, input: bool) -> Wait:
return Wait.any_of(
retry_channel.for_one(),
skip_channel.for_one(),
)
def execute(self, context: Context, input: bool) -> StepDecision:
if retry_channel.results(context):
return go_to(DoWorkStep, False)
return force_fail("manual recovery skipped")
Example: examples/python/dex_examples/patterns/intervention/manual_recovery_flow.py