Deal DSL
Define a durable deal as data, then run the same interpreter for any sellable item.
Product requirements
The process represents one buyer's deal for one item. The item can be a dataset, a physical product, a subscription, a service, or another sellable good. It must:
- Accept a definition containing a process ID, item ID and name, initial state, initial state data, states, actions, waits, and transitions.
- Validate the definition before execution. Every referenced state and action must exist, identifiers must be valid, and each condition name must identify one wait point.
- Run InitializeDeal once. It stores an immutable definition snapshot, buyer and item identity, initial state data, and the first state to enter.
- Run WaitForDealCondition before entering a state. If the state declares a precondition, the Step waits for one message on that condition's keyed Channel. Otherwise it proceeds immediately.
- Merge the condition message into DealStateData, then execute the state's actions in their declared order. Each action gets its own ExecuteDealAction execution and durable checkpoint.
- Run EvaluateDealTransition after the actions. It can wait for another external condition, compare one state-data value against ordered equality cases, and select the matching state or the fallback state.
- Repeat the interpreter loop until it reaches a state without a transition. The Flow then completes with the accumulated state data.
- Keep running executions independent from later definition edits. The Go application stores editable definitions in PostgreSQL and passes a snapshot into each new Flow.
The Flow waits for precondition messages, transition messages, and retry delays after a failed Step attempt. A wait does not occupy a Worker.
Potential failures include:
- Definition validation can reject a missing item, an unknown initial or destination state, a duplicate condition name, an invalid identifier, or an unregistered action.
- A condition message can be malformed, target an unknown or stopped Flow, or arrive for a condition that the definition does not expose.
- A condition result is invalid unless exactly one message satisfies the keyed wait. The Step fails instead of merging ambiguous input.
- Payment, refund, sample delivery, or item delivery can time out or return an error. Dex retries the failed Step without rerunning earlier completed actions.
- An action can succeed remotely while its response is lost. Production adapters should use the Flow ID and Step execution identity as an idempotency key.
- State data can omit a decision key. The transition then follows its explicit fallback state.
- A definition can contain a business loop with no eventual external decision. The Flow remains durable, but product-level validation or timeouts must bound unwanted loops.
Flow design
This interactive definition graph is generated from the runnable Python Flow.
The Flow is a small durable interpreter. A definition chooses the business path; four Step types provide the execution machinery:
DealStart(definition snapshot, buyer, item)
│
▼
InitializeDeal
│
▼
WaitForDealCondition ◄──────────────┐
wait on keyed Channel │
│ │
▼ │
ExecuteDealAction │
one execution per ordered action │
│ │
▼ │
EvaluateDealTransition │
optional wait + case / fallback │
│ │
next state? ── yes ─────────────┘
│ no
▼
complete with accumulated state data
It combines these design patterns:
- Interpreter keeps the executable Flow stable while process definitions supply states, waits, actions, and transitions.
- Finite-state machine makes the current state and every transition explicit and inspectable.
- Immutable execution snapshot prevents an edit to the catalog definition from changing an in-flight deal.
- Keyed Channel rendezvous maps each named external condition to its own durable inbox without creating one Channel definition per condition.
- One-action checkpoint isolates retries and records the exact action that failed. Ordered actions never collapse into one opaque Step.
- Guarded transition with fallback evaluates ordered equality cases against shared state data and always has an explicit default path.
- Blackboard state lets condition messages and actions contribute small key-value updates to one durable DealStateData map.
- Entity store gives each Go execution a stable Flow ID and searchable process, item, buyer, current-state, and pending-condition Attributes.
The portable SDK examples use one action list per state. The Go product example extends the same loop with separate pre-actions and post-actions, a PostgreSQL definition catalog, search endpoints, and a browser UI.
Core implementation
Each tab includes the Flow signature, persistence schema, and the complete four-Step interpreter path. Every snippet comes from the linked runnable example.
class DealDSLFlow(Flow[DealStart]):
definition = Attribute("DealDefinition", DealDefinition)
state_data = Attribute("DealStateData", dict[str, str])
process_id = Attribute("DealProcessID", str)
item_id = Attribute("DealItemID", str)
buyer_id = Attribute("DealBuyerID", str)
current_state = Attribute("DealCurrentState", str)
pending_condition = Attribute("DealPendingCondition", str)
condition_messages = ChannelMap("DealConditionMessages", dict[str, str])
def __init__(self) -> None:
self.initialize = InitializeDeal(self)
self.wait_for_condition = WaitForDealCondition(self)
self.execute_action_step = ExecuteDealAction(self)
self.evaluate_transition = EvaluateDealTransition(self)
def get_steps(self) -> StepList[DealStart]:
return StepList.start_step(self.initialize).other_steps(
self.wait_for_condition,
self.execute_action_step,
self.evaluate_transition,
)
def get_persistence_schema(self) -> PersistenceSchema:
return PersistenceSchema.of(
self.definition,
self.state_data,
self.process_id,
self.item_id,
self.buyer_id,
self.current_state,
self.pending_condition,
self.condition_messages,
)
class InitializeDeal(Step[DealStart]):
def execute(self, context: Context, input: DealStart) -> StepDecision:
definition = input.definition
definition.state(definition.initial_state)
self.flow.definition.set(context, definition)
self.flow.process_id.set(context, definition.process_id)
self.flow.item_id.set(context, definition.item_id)
self.flow.buyer_id.set(context, input.buyer_id)
self.flow.state_data.set(context, dict(definition.initial_state_data))
return go_to(WaitForDealCondition, StateStepInput(definition.initial_state))
class WaitForDealCondition(Step[StateStepInput]):
def wait_for(self, context: Context, input: StateStepInput) -> Wait:
state = self.flow.definition.get(context).state(input.state_name)
if state.pre_condition is None:
return Wait.skip_immediately()
self.flow.pending_condition.set(context, state.pre_condition.name)
return Wait.until(self.flow.condition_messages.for_one(state.pre_condition.name))
def execute(self, context: Context, input: StateStepInput) -> StepDecision:
state = self.flow.definition.get(context).state(input.state_name)
if state.pre_condition is not None:
self.flow.merge_condition(context, state.pre_condition.name)
self.flow.pending_condition.delete(context)
self.flow.current_state.set(context, state.name)
if state.actions:
return go_to(ExecuteDealAction, ActionStepInput(state.name, 0))
return go_to(EvaluateDealTransition, input)
class ExecuteDealAction(Step[ActionStepInput]):
def execute(self, context: Context, input: ActionStepInput) -> StepDecision:
state = self.flow.definition.get(context).state(input.state_name)
if input.action_index < 0 or input.action_index >= len(state.actions):
raise ValueError(f"invalid action index {input.action_index}")
self.flow.execute_action(context, state.actions[input.action_index])
next_index = input.action_index + 1
if next_index < len(state.actions):
return go_to(ExecuteDealAction, ActionStepInput(state.name, next_index))
return go_to(EvaluateDealTransition, StateStepInput(state.name))
class EvaluateDealTransition(Step[StateStepInput]):
def wait_for(self, context: Context, input: StateStepInput) -> Wait:
transition = self.flow.definition.get(context).state(input.state_name).transition
if transition is None or transition.wait_for is None:
return Wait.skip_immediately()
self.flow.pending_condition.set(context, transition.wait_for.name)
return Wait.until(self.flow.condition_messages.for_one(transition.wait_for.name))
def execute(self, context: Context, input: StateStepInput) -> StepDecision:
transition = self.flow.definition.get(context).state(input.state_name).transition
if transition is None:
return graceful_complete(self.flow.state_data.get(context))
if transition.wait_for is not None:
self.flow.merge_condition(context, transition.wait_for.name)
self.flow.pending_condition.delete(context)
state_data = self.flow.state_data.get(context)
next_state = transition.else_state
for deal_case in transition.cases:
if state_data.get(transition.key) == deal_case.equals:
next_state = deal_case.go_to_state
break
return go_to(WaitForDealCondition, StateStepInput(next_state))
Example: examples/python/dex_examples/products/deal_dsl/deal_dsl_flow.py
Demo
This completed Go run sold the Premium research package item to buyer-demo. The Flow waited for buyer-confirmation, charged the buyer, delivered the item, and completed with itemDeliveryStatus: delivered. Each action appears as a separate completed ExecuteDealAction execution.
