Step 基础功能
本页介绍 Step 的默认形态：typed input、可选的 WaitFor、必需的 Execute、通过 StepDecision 做 movement，以及默认 StepOptions 行为。
Dex Step 由原生代码实现。实现该 Step 的 struct 或 class 决定 StepType。在大多数语言里，SDK 默认从原生类型名推导 StepType；你也可以显式覆盖。
同一个 Step type 可以在同一个 Flow 中执行多次。每次执行都是一个 StepExecution，对应一个 StepExecutionID，格式是 StepType-Number，由 Dex 分配并维护。
Step input
起始 Step 接收 Flow input。之后每个 Step 接收调度它的 movement 带来的 input。Step input 随图传递，不是隐藏的全局变量。
def execute(self, context: Context, input: int) -> StepDecision:
return go_to(StepSecond, input + 1)
def execute(self, context: Context, input: int) -> StepDecision:
return graceful_complete(input + 1)
例子: examples/python/dex_examples/primitives/step/step_flow.py
WaitFor and Execute
Step implementation 通常只有一个或两个小方法：WaitFor -> Execute。
WaitFor 是可选的。它向 Dex Server 返回 durable Wait。Dex 持久化该 waiting state，代表应用持续评估这些条件；条件满足后再调用 Execute。
如果 Step 没有实现 WaitFor，则会直接调用 Execute。
Execute 是必需的。它返回 StepDecision，用于流转到下一个 Step、分支到多个 Step，或关闭 Flow。
每次 WaitFor 或 Execute 调用都有独立的 commit 边界。Dex Server 会暂存该 method 内的 Attribute 写入和 Channel publish，并只在 method 成功返回时，将它们与相应的 Wait 或 StepDecision 一起 commit。如果 method 失败，这些 durable 变更都不会被 commit。WaitFor 和 Execute 不共享一次 commit，外部 API 调用也不在这个原子边界内。
在最终 Wait 或 StepDecision 之前，handler 可以发送 heartbeat checkpoint 和任意多条 best-effort Stream message。Response stream 必须以且仅以一个最终 result 结束。这些 progress frame 不属于 method commit，Stream Store 写入失败也不会让 Step 失败。
approval = Channel("Approval", str)
class StepSecond(Step[int]):
def execute(self, context: Context, input: int) -> StepDecision:
return graceful_complete(input + 1)
class ExampleStep(Step[int]):
def __init__(self, second: StepSecond) -> None:
self.second = second
def wait_for(self, context: Context, input: int) -> Wait:
return Wait.until(approval.for_one())
def execute(self, context: Context, input: int) -> StepDecision:
return go_to(StepSecond, input + 1)
例子: examples/python/dex_examples/primitives/step/step_flow.py
Until、AnyOf 和 AllOf 接受没有 condition ID 的 Condition。不要仅仅因为 Condition 嵌套在这些等待组合中就为它添加 ID。AnyCombinationOf 是例外：其中引用的 每个 Condition 都需要唯一的 condition ID。
Step 可等待的 Condition 见 Timer、Channel 与 SubFlow。
StepDecision（movement）
Execute 返回 StepDecision——指向一个或多个后续 Step 的 movement，或关闭决策。一次决策可指向多个 Step，用于表达分支、循环与并行。
见 并行 Step 与 并行 SubFlow 模式。
关闭决策、取消、Flow 输出与 forceCompleteIfChannelsEmpty 见 Step decision。超时、retry 及相关策略的自定义见 Step options。
StepOptions
大多数时候，需要自定义 failure handling —— timeout、retry 与 failure policy。还有 heartbeat、durability 等更高级的 Step options。
- Timeout 自定义一次方法（WaitFor/Execute）调用 attempt 的 timeout。
- Retry policy — 在同一 Step execution 上重复调用同一方法，直到 attempt 耗尽。
- Failure policy — 决定 retry 结束之后发生什么。默认失败 Flow，但 WaitFor 仍可运行 Execute，Execute 也可以前往 recovery Step。
通过在 Step 方法中实现返回 StepOptions 来配置。
def get_options(self) -> StepOptions:
return StepOptions(
execute_retry=RetryPolicy(
initial_interval=timedelta(seconds=1),
backoff_coefficient=2,
maximum_attempts=5,
),
)
def execute(self, context, ready_after_attempt: int) -> StepDecision:
if context.attempt < ready_after_attempt:
raise RuntimeError(f"not ready on attempt {context.attempt}")
return graceful_complete("ready")
例子: examples/python/dex_examples/primitives/step/retry_flow.py
WaitFor failure policy
WaitFor method retry 耗尽时：
- FailFlow — 失败 Flow。这是默认行为。
- PROCEED — 继续调用 Execute。
使用 PROCEED 时，同一 Step 的 Execute 收到不变的 durable input。waitForMethodFailed() 返回 true。getRecoveryError() 包含最后一次失败 WaitFor attempt 的 error_type 和 detail。
def get_step_options(self) -> StepOptions:
return StepOptions(
wait_for_retry=RetryPolicy(maximum_attempts=2),
wait_for_failure=WaitForFailurePolicy.PROCEED,
)
def wait_for(self, context: Context, input: str) -> Wait:
raise RuntimeError("planned WaitFor failure")
def execute(self, context: Context, input: str) -> StepDecision:
if not context.wait_for_method_failed():
raise RuntimeError("waitFor failure was not reported")
return go_to(FinishStep, f"{input}_recovered")
例子: examples/python/dex_examples/primitives/proceed_on_wait_failure/proceed_on_wait_failure_flow.py
Execute failure policy
Execute method retry 耗尽时，onExecuteFailureProceedTo / ExecuteFailure.proceedTo(...) 前往带有相同 input 的 recovery Step。
def get_step_options(self) -> StepOptions:
return StepOptions(
execute_retry=RetryPolicy(maximum_attempts=5)
).on_execute_failure_proceed_to(UpdateQuantityRecovery)
例子: examples/python/dex_examples/patterns/recovery/failure_recovery_flow.py
见 Failure Handling 模式与 money-transfer 用例。
recovery Step 收到失败 Step 不变的 durable input。getRecoveryError() 包含最后一次失败 Execute attempt 的 error_type 和 detail。
Step Failed 事件与 last failure info
- 终态事件 StepExecuteFailed / StepWaitForFailed 在 Step 最终失败时携带 failure output。
- Retry 期间（failure policy 运行前），Step 事件与 live Flow 状态暴露 lastFailureInfo——最近一次 attempt 的错误（type、detail、worker 元数据）。
- 在 Dex Web 中打开事件详情，可在 retry 进行中查看 Last failure。Flow 概览中的活跃 Step 也显示 lastFailureInfo。
Last failure info 描述 retry attempt。Failure policy 决定 retry 结束后的行为。