State Machine: Object Lifecycle Management and State-Driven Business Processes
Business systems are full of "states" — orders have states, approvals have states, rule deployments have states, model training has states. Many teams manage state transitions with if-else or switch-case, and the code quickly becomes unmaintainable:
CoomiaPublished on December 19, 202513 min read
Share this articleTwitter / X
State Machine: Object Lifecycle Management and State-Driven Business Processes
“Series: S10 Design Patterns · Article 5 | Level: Advanced | Reading Time: 18 min
#TL;DR
- Finite State Machines (FSM) are the classic pattern for managing object lifecycles. In coomia-dip, every Ontology object can have a state machine definition — state transitions represent business process progression.
- coomia-dip embeds state machines into ObjectType Schema definitions. Transitions are triggered by Actions, transition conditions are validated by Guard functions, and side effects are executed through event-driven mechanisms.
- Combined with the Saga pattern and Event Sourcing, state machines provide auditable, replayable, and visualizable business process management.
#Introduction: Why State Machines Are Needed
Business systems are full of "states" — orders have states, approvals have states, rule deployments have states, model training has states. Many teams manage state transitions with if-else or switch-case, and the code quickly becomes unmaintainable:
Python
# Anti-pattern: scattered state checks
def approve_order(order):
if order.status == "pending":
if order.amount > 10000:
if has_manager_approval(order):
order.status = "approved"
else:
order.status = "pending_manager"
else:
order.status = "approved"
elif order.status == "pending_manager":
if is_manager(current_user):
order.status = "approved"
else:
raise ValueError("Not authorized")
elif order.status == "approved":
raise ValueError("Already approved")
# ... more and more branches
The problem: state transition logic is scattered across business code with no global view and no way to answer the basic question "what states can this object transition from and to."
The state machine pattern centralizes all states and transitions, providing a clear global view and strict transition constraints.
#Part 1: Core State Machine Model
#1.1 coomia-dip State Machine Definition
coomia-dip defines state machines at the Ontology Schema level, making them first-class citizens of ObjectTypes:
Python
from dataclasses import dataclass, field
from enum import Enum
from typing import Any, Callable, Awaitable
@dataclass(frozen=True)
class State:
"""A state in the state machine."""
name: str
display_name: str
description: str
is_initial: bool = False
is_terminal: bool = False
metadata: dict[str, Any] = field(default_factory=dict)
@dataclass(frozen=True)
class Transition:
"""A transition between states."""
name: str
from_state: str
to_state: str
action: str # Action that triggers this transition
guard: str | None = None # Guard condition expression
side_effects: list[str] = field(default_factory=list)
required_permissions: list[str] = field(default_factory=list)
metadata: dict[str, Any] = field(default_factory=dict)
@dataclass
class StateMachineDefinition:
"""Complete state machine definition for an ObjectType."""
object_type: str
states: list[State]
transitions: list[Transition]
initial_state: str
history_enabled: bool = True
version: int = 1
def get_available_transitions(self, current_state: str) -> list[Transition]:
"""Get all transitions available from the current state."""
return [t for t in self.transitions if t.from_state == current_state]
def validate(self) -> list[str]:
"""Validate the state machine definition."""
errors: list[str] = []
state_names = {s.name for s in self.states}
if self.initial_state not in state_names:
errors.append(f"Initial state '{self.initial_state}' not found")
for t in self.transitions:
if t.from_state not in state_names:
errors.append(f"Transition '{t.name}': from_state '{t.from_state}' not found")
if t.to_state not in state_names:
errors.append(f"Transition '{t.name}': to_state '{t.to_state}' not found")
reachable = {self.initial_state}
changed = True
while changed:
changed = False
for t in self.transitions:
if t.from_state in reachable and t.to_state not in reachable:
reachable.add(t.to_state)
changed = True
unreachable = state_names - reachable
if unreachable:
errors.append(f"Unreachable states: {unreachable}")
terminals = [s for s in self.states if s.is_terminal]
if not terminals:
errors.append("No terminal states defined")
return errors
#1.2 Guard Conditions
Guards are validation functions executed before state transitions. A transition is only permitted when the Guard returns True:
Python
class GuardRegistry:
"""Registry for state machine guard functions."""
def __init__(self):
self._guards: dict[str, Callable[..., Awaitable[bool]]] = {}
def register(self, name: str, guard_fn: Callable[..., Awaitable[bool]]) -> None:
"""Register a guard function."""
self._guards[name] = guard_fn
async def evaluate(
self,
guard_name: str,
obj: dict[str, Any],
context: "TransitionContext",
) -> bool:
"""Evaluate a guard condition."""
guard_fn = self._guards.get(guard_name)
if guard_fn is None:
raise ValueError(f"Guard '{guard_name}' not registered")
return await guard_fn(obj, context)
guard_registry = GuardRegistry()
async def amount_below_threshold(obj: dict, ctx: "TransitionContext") -> bool:
"""Guard: order amount below auto-approval threshold."""
return obj.get("amount", 0) <= ctx.config.get("auto_approve_threshold", 10000)
async def has_required_approvals(obj: dict, ctx: "TransitionContext") -> bool:
"""Guard: all required approvals have been received."""
required = obj.get("required_approvers", [])
received = obj.get("approvals", [])
return all(a in received for a in required)
async def data_quality_passed(obj: dict, ctx: "TransitionContext") -> bool:
"""Guard: data quality checks have passed."""
return obj.get("quality_score", 0) >= ctx.config.get("min_quality_score", 0.95)
guard_registry.register("amount_below_threshold", amount_below_threshold)
guard_registry.register("has_required_approvals", has_required_approvals)
guard_registry.register("data_quality_passed", data_quality_passed)
#1.3 State Machine Execution Engine
Python
@dataclass
class TransitionContext:
"""Context for a state transition."""
actor_id: str
actor_type: str
tenant_id: str
world_id: str
timestamp: datetime
config: dict[str, Any] = field(default_factory=dict)
metadata: dict[str, Any] = field(default_factory=dict)
@dataclass
class TransitionResult:
"""Result of a state transition attempt."""
success: bool
from_state: str
to_state: str | None
transition_name: str | None
error: str | None = None
side_effect_results: list[dict] = field(default_factory=list)
class StateMachineEngine:
"""Engine for executing state machine transitions."""
def __init__(
self,
definition: StateMachineDefinition,
guard_registry: GuardRegistry,
event_store: "EventStore",
side_effect_executor: "SideEffectExecutor",
):
self._definition = definition
self._guards = guard_registry
self._event_store = event_store
self._side_effects = side_effect_executor
async def transition(
self,
obj: dict[str, Any],
action: str,
context: TransitionContext,
) -> TransitionResult:
"""Attempt a state transition triggered by an action."""
current_state = obj.get("_state", self._definition.initial_state)
matching = [
t for t in self._definition.transitions
if t.from_state == current_state and t.action == action
]
if not matching:
return TransitionResult(
success=False,
from_state=current_state,
to_state=None,
transition_name=None,
error=f"No transition for action '{action}' from state '{current_state}'",
)
for transition in matching:
if transition.guard:
guard_passed = await self._guards.evaluate(
transition.guard, obj, context
)
if not guard_passed:
continue
if transition.required_permissions:
if not await self._check_permissions(
context.actor_id, transition.required_permissions
):
return TransitionResult(
success=False,
from_state=current_state,
to_state=transition.to_state,
transition_name=transition.name,
error="Insufficient permissions",
)
obj["_state"] = transition.to_state
obj["_state_changed_at"] = context.timestamp.isoformat()
obj["_state_changed_by"] = context.actor_id
await self._event_store.append([
DomainEvent(
event_id=generate_id(),
event_type="state_machine.transition",
aggregate_id=obj["_id"],
aggregate_type=self._definition.object_type,
sequence_number=await self._next_sequence(obj["_id"]),
timestamp=context.timestamp,
payload={
"from_state": current_state,
"to_state": transition.to_state,
"transition": transition.name,
"action": action,
},
metadata=EventMetadata(
actor_id=context.actor_id,
actor_type=context.actor_type,
tenant_id=context.tenant_id,
world_id=context.world_id,
source_plane="control",
trace_id=generate_trace_id(),
),
)
])
se_results = []
for se_name in transition.side_effects:
result = await self._side_effects.execute(se_name, obj, context)
se_results.append(result)
return TransitionResult(
success=True,
from_state=current_state,
to_state=transition.to_state,
transition_name=transition.name,
side_effect_results=se_results,
)
return TransitionResult(
success=False,
from_state=current_state,
to_state=None,
transition_name=None,
error="All guard conditions failed",
)
async def _check_permissions(self, actor_id: str, required: list[str]) -> bool:
return True
async def _next_sequence(self, aggregate_id: str) -> int:
return 0
#Part 2: Practical Object Lifecycle Management
#2.1 Risk Rule Lifecycle
Using a risk control rule as an example, here is its complete lifecycle state machine:
Python
risk_rule_state_machine = StateMachineDefinition(
object_type="RiskRule",
initial_state="draft",
states=[
State("draft", "Draft", "Rule is being edited", is_initial=True),
State("review", "Under Review", "Rule submitted for review"),
State("testing", "Testing", "Rule being tested in sandbox"),
State("approved", "Approved", "Rule passed review"),
State("deploying", "Deploying", "Rule being deployed to production"),
State("active", "Active", "Rule running in production"),
State("suspended", "Suspended", "Rule temporarily suspended"),
State("deprecated", "Deprecated", "Rule deprecated"),
State("archived", "Archived", "Rule archived", is_terminal=True),
],
transitions=[
Transition("submit_review", "draft", "review", "submit",
guard="has_required_fields",
side_effects=["notify_reviewers"]),
Transition("approve", "review", "approved", "approve",
guard="has_required_approvals",
required_permissions=["rule.approve"],
side_effects=["notify_submitter"]),
Transition("reject", "review", "draft", "reject",
side_effects=["notify_submitter_rejection"]),
Transition("start_test", "approved", "testing", "test",
side_effects=["create_sandbox_instance"]),
Transition("test_passed", "testing", "approved", "pass_test",
guard="data_quality_passed",
side_effects=["record_test_results"]),
Transition("test_failed", "testing", "draft", "fail_test",
side_effects=["record_test_failures"]),
Transition("deploy", "approved", "deploying", "deploy",
required_permissions=["rule.deploy"],
side_effects=["trigger_deployment_saga"]),
Transition("deployment_complete", "deploying", "active", "activate",
side_effects=["notify_stakeholders"]),
Transition("deployment_failed", "deploying", "approved", "rollback",
side_effects=["notify_ops_team"]),
Transition("suspend", "active", "suspended", "suspend",
required_permissions=["rule.suspend"],
side_effects=["disable_rule_execution"]),
Transition("resume", "suspended", "active", "resume",
required_permissions=["rule.resume"],
side_effects=["enable_rule_execution"]),
Transition("deprecate", "active", "deprecated", "deprecate",
side_effects=["notify_dependents"]),
Transition("deprecate_suspended", "suspended", "deprecated", "deprecate",
side_effects=["notify_dependents"]),
Transition("archive", "deprecated", "archived", "archive",
side_effects=["cleanup_resources"]),
],
)
#2.2 State Machine Visualization
The coomia-dip Platform Console automatically generates visual state transition diagrams from StateMachineDefinitions:
Python
class StateMachineVisualizer:
"""Generate visual representations of state machines."""
def to_mermaid(self, definition: StateMachineDefinition) -> str:
"""Generate Mermaid state diagram."""
lines = ["stateDiagram-v2"]
lines.append(f" [*] --> {definition.initial_state}")
for t in definition.transitions:
label = t.name
if t.guard:
label += f" [{t.guard}]"
lines.append(f" {t.from_state} --> {t.to_state}: {label}")
for s in definition.states:
if s.is_terminal:
lines.append(f" {s.name} --> [*]")
return "\n".join(lines)
def to_dot(self, definition: StateMachineDefinition) -> str:
"""Generate Graphviz DOT representation."""
lines = [
"digraph StateMachine {",
" rankdir=LR;",
' node [shape=box, style=rounded];',
]
for s in definition.states:
attrs = []
if s.is_initial:
attrs.append("peripheries=2")
if s.is_terminal:
attrs.append("shape=doublecircle")
attr_str = f" [{', '.join(attrs)}]" if attrs else ""
lines.append(f' {s.name} [label="{s.display_name}"{attr_str}];')
for t in definition.transitions:
label = t.action
if t.guard:
label += f"\\n[{t.guard}]"
lines.append(f' {t.from_state} -> {t.to_state} [label="{label}"];')
lines.append("}")
return "\n".join(lines)
#2.3 State History Queries
With history_enabled, you can query an object's complete state change history:
Python
class StateHistoryService:
"""Service for querying state transition history."""
async def get_history(
self, object_type: str, object_id: str
) -> list[dict[str, Any]]:
"""Get complete state transition history for an object."""
events = await self._event_store.get_events(
aggregate_id=object_id,
aggregate_type=object_type,
event_type="state_machine.transition",
)
return [
{
"from_state": e.payload["from_state"],
"to_state": e.payload["to_state"],
"transition": e.payload["transition"],
"action": e.payload["action"],
"actor": e.metadata.actor_id,
"timestamp": e.timestamp.isoformat(),
}
for e in events
]
async def get_state_at(
self, object_type: str, object_id: str, timestamp: datetime
) -> str:
"""Get the state of an object at a specific point in time."""
events = await self._event_store.get_events(
aggregate_id=object_id,
aggregate_type=object_type,
event_type="state_machine.transition",
before=timestamp,
)
if not events:
definition = await self._get_definition(object_type)
return definition.initial_state
return events[-1].payload["to_state"]
async def get_dwell_time(
self, object_type: str, object_id: str, state: str
) -> timedelta:
"""Calculate how long an object has been in a specific state."""
events = await self._event_store.get_events(
aggregate_id=object_id,
aggregate_type=object_type,
event_type="state_machine.transition",
)
total = timedelta()
entered_at: datetime | None = None
for e in events:
if e.payload["to_state"] == state:
entered_at = e.timestamp
elif e.payload["from_state"] == state and entered_at:
total += e.timestamp - entered_at
entered_at = None
if entered_at:
total += datetime.utcnow() - entered_at
return total
#Part 3: Hierarchical State Machines
#3.1 Nested States
For complex business processes, coomia-dip supports hierarchical state machines — states can contain sub-state machines:
Python
@dataclass
class HierarchicalState(State):
"""A state that contains a sub-state machine."""
sub_machine: StateMachineDefinition | None = None
testing_sub_machine = StateMachineDefinition(
object_type="RiskRule.testing",
initial_state="unit_test",
states=[
State("unit_test", "Unit Test", "Executing unit tests", is_initial=True),
State("integration_test", "Integration Test", "Executing integration tests"),
State("performance_test", "Performance Test", "Executing performance tests"),
State("sandbox_validation", "Sandbox Validation", "Sandbox environment validation"),
State("test_complete", "Test Complete", "All tests passed", is_terminal=True),
],
transitions=[
Transition("unit_pass", "unit_test", "integration_test", "pass_unit"),
Transition("integration_pass", "integration_test", "performance_test", "pass_integration"),
Transition("perf_pass", "performance_test", "sandbox_validation", "pass_performance"),
Transition("sandbox_pass", "sandbox_validation", "test_complete", "pass_sandbox"),
],
)
#3.2 Parallel States
Some scenarios require parallel state regions, such as an approval process needing both legal and technical review simultaneously:
Python
@dataclass
class ParallelRegion:
"""A parallel region in a state machine."""
name: str
sub_machine: StateMachineDefinition
required: bool = True
@dataclass
class ParallelState(State):
"""A state with parallel regions."""
regions: list[ParallelRegion] = field(default_factory=list)
join_condition: str = "all" # all | any
legal_review = StateMachineDefinition(
object_type="RiskRule.review.legal",
initial_state="pending",
states=[
State("pending", "Pending", "", is_initial=True),
State("approved", "Approved", "", is_terminal=True),
State("rejected", "Rejected", "", is_terminal=True),
],
transitions=[
Transition("approve", "pending", "approved", "legal_approve"),
Transition("reject", "pending", "rejected", "legal_reject"),
],
)
tech_review = StateMachineDefinition(
object_type="RiskRule.review.tech",
initial_state="pending",
states=[
State("pending", "Pending", "", is_initial=True),
State("approved", "Approved", "", is_terminal=True),
State("rejected", "Rejected", "", is_terminal=True),
],
transitions=[
Transition("approve", "pending", "approved", "tech_approve"),
Transition("reject", "pending", "rejected", "tech_reject"),
],
)
#Part 4: State Machine and Saga Collaboration
#4.1 State Transitions Triggering Sagas
When a state machine transition involves cross-Layer operations, the transition's side effect triggers a Saga:
Python
class DeploymentSideEffect:
"""Side effect that triggers a deployment Saga."""
async def execute(
self, obj: dict[str, Any], context: TransitionContext
) -> dict:
"""Trigger deployment Saga when rule transitions to 'deploying'."""
saga = CreateRiskRuleSaga().build({
"schema": obj["schema"],
"storage": obj["storage_config"],
"reasoning": obj["reasoning_config"],
"monitoring": obj["monitoring_config"],
})
result = await self._saga_orchestrator.execute(saga)
if result.status == "completed":
await self._state_engine.transition(obj, "activate", context)
else:
await self._state_engine.transition(obj, "rollback", context)
return {"saga_id": saga.saga_id, "status": result.status}
#4.2 Saga Steps as State Transitions
Conversely, each Saga step's execution can be modeled as a state transition of the Saga instance itself:
Code
Saga State Machine:
CREATED → STEP_1_RUNNING → STEP_1_COMPLETED → STEP_2_RUNNING → ...
↘ STEP_1_FAILED → COMPENSATING → COMPENSATED
#Part 5: Production Practices
#5.1 Concurrency Safety for State Transitions
Multiple users may simultaneously attempt state transitions on the same object. coomia-dip uses optimistic locking for concurrency safety:
Python
class ConcurrentStateMachine:
"""State machine with optimistic concurrency control."""
async def transition(
self, obj_id: str, action: str, context: TransitionContext
) -> TransitionResult:
"""Execute transition with optimistic locking."""
obj = await self._repository.get(obj_id)
current_version = obj["_version"]
result = await self._engine.transition(obj, action, context)
if result.success:
updated = await self._repository.update(
obj_id, obj, expected_version=current_version
)
if not updated:
raise ConcurrencyError(
f"Object {obj_id} was modified concurrently"
)
return result
#5.2 State Transition Timeouts
Objects lingering in intermediate states may indicate problems. coomia-dip supports state timeout configuration:
Python
@dataclass
class StateTimeout:
"""Timeout configuration for a state."""
state: str
timeout: timedelta
timeout_action: str
notification_before: timedelta | None = None
class StateTimeoutMonitor:
"""Monitor for state timeouts."""
async def check_timeouts(self) -> list[dict]:
"""Check all objects for state timeouts."""
timed_out = []
for timeout_config in self._timeout_configs:
objects = await self._repository.find_in_state(
timeout_config.state,
entered_before=datetime.utcnow() - timeout_config.timeout,
)
for obj in objects:
timed_out.append({
"object_id": obj["_id"],
"state": timeout_config.state,
"timeout_action": timeout_config.timeout_action,
})
await self._engine.transition(
obj, timeout_config.timeout_action,
TransitionContext(
actor_id="system",
actor_type="timeout_monitor",
tenant_id=obj["_tenant_id"],
world_id=obj["_world_id"],
timestamp=datetime.utcnow(),
),
)
return timed_out
#5.3 State Analytics
Python
class StateAnalyticsService:
"""Analytics for state machine usage."""
async def get_state_distribution(
self, object_type: str, tenant_id: str
) -> dict[str, int]:
"""Get current distribution of objects across states."""
return await self._repository.count_by_state(object_type, tenant_id)
async def get_bottleneck_states(
self, object_type: str, tenant_id: str
) -> list[dict]:
"""Identify states where objects tend to get stuck."""
distribution = await self.get_state_distribution(object_type, tenant_id)
dwell_times = await self.get_avg_dwell_time(object_type, tenant_id)
bottlenecks = []
for state, count in distribution.items():
avg_dwell = dwell_times.get(state, 0)
if count > 10 and avg_dwell > 86400:
bottlenecks.append({
"state": state,
"count": count,
"avg_dwell_seconds": avg_dwell,
})
return sorted(bottlenecks, key=lambda x: x["count"], reverse=True)
#Part 6: Anti-Patterns and Best Practices
#6.1 Avoid State Explosion
Too many states make state machines hard to understand and maintain. Rules of thumb:
- No more than 12 states in a single-level state machine
- Use hierarchical state machines for complex processes
- Use "metadata" instead of "micro-states" — don't create new states for every minor variation
#6.2 State Transitions Must Be Atomic
A transition must either fully succeed (state change + side effect execution) or fully fail (state unchanged). Never allow "state changed but side effect not executed."
#6.3 Terminal States Must Be Stable
Terminal states mean the object no longer accepts any state transitions. Do not define outgoing transitions on terminal states. If you need to "reactivate" an archived object, create a new object rather than modifying the terminal state.
#Key Takeaways
- First-Class Citizen: State machines are first-class citizens of coomia-dip ObjectType Schemas — state transitions represent business process progression
- Guard + Side Effect: Guard conditions control whether transitions are allowed; side effects execute post-transition actions
- Event-Driven: Every state transition is recorded as an event, supporting audit, replay, and time-travel queries
- Hierarchical: Complex processes are managed through nested state machines and parallel regions
- Saga Collaboration: Cross-Layer state transitions use Sagas to ensure eventual consistency
- Visualization: State transition diagrams are automatically generated from definitions, providing a global view
#Next Article
In the next article, we will explore the Sandbox Pattern — how coomia-dip provides safe, isolated execution environments for rule testing, policy validation, and Agent debugging.
S10-06: Sandbox Pattern: Safe Isolated Execution Environments
#Tags
#DesignPatterns #StateMachine #FSM #Lifecycle #Guard #SideEffect #HierarchicalStateMachine #ParallelStates #ObjectManagement