Menu
>-
Por padrão, está selecionado o prompt que primeiro revisa a origem. Você pode mudar para um comando direto ou baixar uma cópia local.
Leia o SKILL.md e os arquivos complementares exibidos pelo SkillsMP antes de decidir se vai instalar.
Instalar com Codex ou Claude Copie este prompt, cole no Codex, Claude ou outro assistente e deixe que ele revise a página da skill e instale para você.
Um comando direto ignora o prompt de revisão. Verifique a origem antes de executá-lo.
npx skills add https://github.com/CleanExpo/ATO --skill state-machineO comando permanece em uma só linha. Role horizontalmente para revisá-lo antes de copiar.
Prefere uma cópia local? Baixe os arquivos disponíveis atualmente no SkillsMP.
Route complex requests to the right specialist agent or chain of agents. This skill acts as the central brain of an agent swarm — it analyses what the user needs, determines which specialist domain(s) are required, and coordinates parallel or sequential agent execution. Use this skill when a request spans multiple domains (e.g., "research competitors and create a pitch deck"), when you need to decide which specialist should handle an ambiguous request, or when a task requires a multi-step pipeline across different skills. Triggers on: multi-step requests, cross-domain tasks, "coordinate", "plan this out", "I need help with multiple things", or any complex request that touches more than one specialist area. Also triggers when the user seems unsure which tool or approach to use.
>-
>-
Exibindo SKILL.md
Baseado na classificação ocupacional SOC
| id | state-machine |
| name | state-machine |
| type | skill |
| version | 1.0.0 |
| created | 20/03/2026 |
| modified | 20/03/2026 |
| status | active |
| metadata | {"author":"NodeJS-Starter-V1","version":"1.0.0","locale":"en-AU"} |
| description | >- |
| context | fork |
Enforces deterministic, minimal state machines for any multi-step flow in NodeJS-Starter-V1. Codifies the project's existing patterns (TaskStatus, ExecutionStatus, NodeStatus) and provides a reusable framework for defining states, transition maps, guard conditions, and side effects across Python and TypeScript.
Enum status fields to Pydantic modelsis_active, is_published)data-validation instead)scientific-luxury + Framer Motion)error-taxonomy instead)Use str, Enum for all status fields. String enums serialise cleanly to JSON and are readable in logs.
from enum import Enum
class OrderStatus(str, Enum):
"""Status of an order — each value is a distinct FSM state."""
DRAFT = "draft"
SUBMITTED = "submitted"
PROCESSING = "processing"
COMPLETED = "completed"
CANCELLED = "cancelled"
= {
: ,
: ,
: ,
: ,
: ,
} ;
= ( )[keyof ];
Location: apps/backend/src/agents/orchestrator.py
10 states governing the task lifecycle with verification loop:
PENDING → IN_PROGRESS → AWAITING_VERIFICATION
↓
VERIFICATION_IN_PROGRESS
↙ ↘
VERIFICATION_PASSED VERIFICATION_FAILED
↓ ↓
COMPLETED (retry → IN_PROGRESS)
(max retries → ESCALATED_TO_HUMAN)
Cross-cutting:
Any non-terminal → BLOCKED (dependency unmet)
Any non-terminal → FAILED (unrecoverable error)
Terminal states: COMPLETED, FAILED, ESCALATED_TO_HUMAN
Location: apps/backend/src/workflow/models.py
4 states for workflow execution lifecycle:
PENDING → RUNNING → COMPLETED
→ FAILED
Terminal states: COMPLETED, FAILED
Location: apps/backend/src/workflow/state.py
5 states for individual node execution:
PENDING → RUNNING → COMPLETED
→ FAILED
→ SKIPPED (conditional branch not taken)
Terminal states: COMPLETED, FAILED, SKIPPED
Define allowed transitions as a dictionary. Reject any transition not in the map.
from typing import TypeVar
S = TypeVar("S", bound=Enum)
TASK_TRANSITIONS: dict[TaskStatus, set[TaskStatus]] = {
TaskStatus.PENDING: {
TaskStatus.IN_PROGRESS,
TaskStatus.BLOCKED,
TaskStatus.FAILED,
},
TaskStatus.IN_PROGRESS: {
TaskStatus.AWAITING_VERIFICATION,
TaskStatus.BLOCKED,
TaskStatus.FAILED,
},
TaskStatus.AWAITING_VERIFICATION: {
TaskStatus.VERIFICATION_IN_PROGRESS,
},
TaskStatus.VERIFICATION_IN_PROGRESS: {
TaskStatus.VERIFICATION_PASSED,
TaskStatus.VERIFICATION_FAILED,
},
TaskStatus.VERIFICATION_PASSED: {
TaskStatus.COMPLETED,
},
TaskStatus.VERIFICATION_FAILED: {
TaskStatus.IN_PROGRESS, # Retry
TaskStatus.ESCALATED_TO_HUMAN, # Max retries exceeded
},
# Terminal states — no outgoing transitions
TaskStatus.COMPLETED: set(),
TaskStatus.FAILED: set(),
TaskStatus.BLOCKED: {
TaskStatus.PENDING, # Unblocked
},
TaskStatus.ESCALATED_TO_HUMAN: set(),
}
def validate_transition(
current: TaskStatus,
target: TaskStatus,
transitions: dict[TaskStatus, set[TaskStatus]] = TASK_TRANSITIONS,
) -> bool:
"""Check if a state transition is allowed."""
allowed = transitions.get(current, set())
return target in allowed
def transition(
current: TaskStatus,
target: TaskStatus,
) -> TaskStatus:
"""Execute a validated state transition."""
if not validate_transition(current, target):
raise ValueError(
f"Invalid transition: {current.value} → {target.value}"
)
return target
Guards are predicates that must be true before a transition fires.
from dataclasses import dataclass
from typing import Callable
@dataclass
class GuardedTransition:
"""A transition with a guard condition."""
source: TaskStatus
target: TaskStatus
guard: Callable[[TaskState], bool]
description: str
GUARDED_TRANSITIONS = [
GuardedTransition(
source=TaskStatus.VERIFICATION_FAILED,
target=TaskStatus.IN_PROGRESS,
guard=lambda task: task.attempts < task.max_attempts,
description="Retry only if under max attempts",
),
GuardedTransition(
source=TaskStatus.VERIFICATION_FAILED,
target=TaskStatus.ESCALATED_TO_HUMAN,
guard=lambda task: task.attempts >= task.max_attempts,
description="Escalate after max retry attempts",
),
GuardedTransition(
source=TaskStatus.BLOCKED,
target=TaskStatus.PENDING,
guard=lambda task: len(task.error_history) == 0,
description="Unblock only if blocking condition resolved",
),
]
Execute side effects (logging, notifications, metrics) when transitions occur.
import structlog
logger = structlog.get_logger(__name__)
async def on_transition(
task: TaskState,
old_status: TaskStatus,
new_status: TaskStatus,
) -> None:
"""Execute side effects after a state transition."""
logger.info(
"task_transition",
task_id=task.task_id,
from_status=old_status.value,
to_status=new_status.value,
)
# Terminal state side effects
if new_status == TaskStatus.COMPLETED:
task.completed_at = datetime.now().isoformat()
if new_status == TaskStatus.ESCALATED_TO_HUMAN:
logger.warning(
"task_escalated",
task_id=task.task_id,
attempts=task.attempts,
)
if new_status == TaskStatus.FAILED:
logger.error(
"task_failed",
task_id=task.task_id,
error_history=task.error_history,
)
Use a reducer pattern for UI state machines:
type FetchState =
| { status: 'idle' }
| { status: 'loading' }
| { status: 'success'; data: unknown }
| { status: 'error'; error: string };
type FetchEvent =
| { type: 'FETCH' }
| { type: 'RESOLVE'; data: unknown }
| { type: 'REJECT'; error: string }
| { type: 'RESET' };
function fetchReducer(state: FetchState, event: FetchEvent): FetchState {
switch (state.status) {
case 'idle':
if (event.type === 'FETCH') return { status: 'loading' };
return state;
case 'loading':
if (event.type === 'RESOLVE') return { status: 'success', data: event.data };
if (event.type === 'REJECT') return { status: 'error', error: event.error };
return state;
case 'success':
if (event.type === 'FETCH') return { status: 'loading' };
if (event.type === 'RESET') return { status: 'idle' };
return state;
case 'error':
if (event.type === 'FETCH') return { status: 'loading' };
if (event.type === 'RESET') return { status: 'idle' };
return state;
}
}
Usage with React:
function useStateMachine() {
const [state, dispatch] = useReducer(fetchReducer, { status: 'idle' });
return { state, dispatch };
}
List every distinct status. Apply the Turing Check — no redundant states:
class MyStatus(str, Enum):
"""Each state must be reachable and have at least one exit or be terminal."""
STATE_A = "state_a"
STATE_B = "state_b"
STATE_C = "state_c" # Terminal
MY_TRANSITIONS: dict[MyStatus, set[MyStatus]] = {
MyStatus.STATE_A: {MyStatus.STATE_B},
MyStatus.STATE_B: {MyStatus.STATE_C},
MyStatus.STATE_C: set(), # Terminal
}
Only add guards when a transition depends on runtime data.
Logging, metrics, notifications on transition.
def validate_fsm(transitions: dict[Enum, set[Enum]]) -> list[str]:
"""Validate a finite state machine for completeness."""
issues: list[str] = []
all_states = set(transitions.keys())
all_targets = {t for targets in transitions.values() for t in targets}
# Check for unreachable states (not targets and not initial)
initial = list(transitions.keys())[0]
unreachable = all_states - all_targets - {initial}
for state in unreachable:
issues.append(f"Unreachable state: {state.value}")
# Check for undefined targets
undefined = all_targets - all_states
for state in undefined:
issues.append(f"Undefined target state: {state.value}")
# Check terminal states have no transitions
for state, targets in transitions.items():
if len(targets) == 0:
continue # Valid terminal
# Non-terminal must have at least one reachable target
if not targets.intersection(all_states):
issues.append(f"Dead end: {state.value} targets undefined states")
return issues
| Anti-Pattern | Why It Fails | Correct Approach |
|---|---|---|
String literals for status (status = "running") | No type safety, typos cause bugs | str, Enum with exhaustive matching |
Boolean flags (is_running, is_complete) | Impossible states (is_running=True, is_complete=True) | Single status enum |
| Implicit transitions (set status anywhere) | Race conditions, invalid states | Transition map with validation |
| God enum (20+ states) | Unmanageable complexity | Decompose into hierarchical machines |
| Missing terminal states | Processes hang forever | Every machine needs at least one terminal |
str, Enumvalidate_transition() called before every status changefrom_status and to_status[AGENT_ACTIVATED]: State Machine
[PHASE]: {Design | Implementation | Review}
[STATUS]: {in_progress | complete}
{state machine analysis or implementation guidance}
[NEXT_ACTION]: {what to do next}
WORKFLOW_VALIDATION_INVALID_TRANSITIONtask_transition event with from_status, to_status