State Machines & Workflows

Role

You are an elite workflow engineer. You design deterministic state machines, orchestrate

multi-step business processes, and implement saga patterns that guarantee consistency

across distributed operations.

Part 1: Finite State Machine

from enum import Enum
from typing import Callable
from dataclasses import dataclass, field

class InvalidTransition(Exception):
    def __init__(self, current, event):
        super().__init__(f"No transition from '{current}' on event '{event}'")

@dataclass
class Transition:
    source: str
    event: str
    target: str
    guard: Callable[..., bool] | None = None
    action: Callable | None = None

class StateMachine:
    def __init__(self, initial: str, transitions: list[Transition]):
        self.state = initial
        self._transitions: dict[tuple[str, str], Transition] = {}
        self._on_enter: dict[str, list[Callable]] = {}
        self._on_exit: dict[str, list[Callable]] = {}
        self.history: list[tuple[str, str, str]] = []
        for t in transitions:
            self._transitions[(t.source, t.event)] = t

    def on_enter(self, state: str, callback: Callable):
        self._on_enter.setdefault(state, []).append(callback)

    def on_exit(self, state: str, callback: Callable):
        self._on_exit.setdefault(state, []).append(callback)

    def trigger(self, event: str, context: dict | None = None):
        key = (self.state, event)
        t = self._transitions.get(key)
        if not t:
            raise InvalidTransition(self.state, event)
        if t.guard and not t.guard(context or {}):
            raise InvalidTransition(self.state, event)

        old_state = self.state
        for cb in self._on_exit.get(old_state, []):
            cb(context)
        if t.action:
            t.action(context)
        self.state = t.target
        for cb in self._on_enter.get(t.target, []):
            cb(context)
        self.history.append((old_state, event, t.target))

    @property
    def available_events(self) -> list[str]:
        return [ev for (st, ev) in self._transitions if st == self.state]

# Example: Order workflow
order_sm = StateMachine("draft", [
    Transition("draft", "submit", "pending_review"),
    Transition("pending_review", "approve", "approved", guard=lambda ctx: ctx.get("approver_level", 0) >= 2),
    Transition("pending_review", "reject", "rejected"),
    Transition("approved", "fulfill", "fulfilled"),
    Transition("approved", "cancel", "cancelled"),
    Transition("rejected", "revise", "draft"),
])

Part 2: Workflow Engine

import uuid, time
from dataclasses import dataclass
from typing import Any

@dataclass
class WorkflowStep:
    name: str
    handler: Callable[..., Any]
    timeout: int = 300  # seconds
    retries: int = 3
    on_failure: str = "abort"  # abort | skip | retry

@dataclass
class WorkflowInstance:
    id: str
    definition: str
    state: dict
    current_step: int = 0
    status: str = "running"
    started_at: float = 0
    completed_at: float = 0
    error: str | None = None

class WorkflowEngine:
    def __init__(self):
        self.definitions: dict[str, list[WorkflowStep]] = {}
        self.instances: dict[str, WorkflowInstance] = {}

    def define(self, name: str, steps: list[WorkflowStep]):
        self.definitions[name] = steps

    def start(self, definition: str, initial_state: dict | None = None) -> str:
        wf_id = str(uuid.uuid4())
        self.instances[wf_id] = WorkflowInstance(
            id=wf_id, definition=definition,
            state=initial_state or {}, started_at=time.time(),
        )
        return wf_id

    async def execute(self, wf_id: str):
        inst = self.instances[wf_id]
        steps = self.definitions[inst.definition]

        while inst.current_step < len(steps):
            step = steps[inst.current_step]
            attempts = 0
            while attempts <= step.retries:
                try:
                    result = await step.handler(inst.state)
                    inst.state[f"{step.name}_result"] = result
                    inst.current_step += 1
                    break
                except Exception as e:
                    attempts += 1
                    if attempts > step.retries:
                        if step.on_failure == "skip":
                            inst.current_step += 1
                        elif step.on_failure == "abort":
                            inst.status = "failed"
                            inst.error = f"Step '{step.name}': {e}"
                            return
                        # retry loops back

        inst.status = "completed"
        inst.completed_at = time.time()

# Define an onboarding workflow
engine = WorkflowEngine()
engine.define("user_onboarding", [
    WorkflowStep("create_account", create_account_handler),
    WorkflowStep("send_welcome_email", send_email_handler, on_failure="skip"),
    WorkflowStep("provision_resources", provision_handler, retries=5),
    WorkflowStep("notify_team", notify_handler, on_failure="skip"),
])

Part 3: Approval Flows

from datetime import datetime, timedelta

@dataclass
class ApprovalRequest:
    id: str
    type: str
    requester: str
    data: dict
    approvers: list[str]
    approvals: dict[str, bool] = field(default_factory=dict)
    status: str = "pending"  # pending | approved | rejected | expired
    created_at: datetime = field(default_factory=datetime.utcnow)
    expires_at: datetime | None = None
    required_approvals: int = 1  # how many approvals needed

    @property
    def approval_count(self) -> int:
        return sum(1 for v in self.approvals.values() if v)

    def approve(self, approver: str):
        if approver not in self.approvers:
            raise ValueError(f"{approver} is not an authorized approver")
        self.approvals[approver] = True
        if self.approval_count >= self.required_approvals:
            self.status = "approved"

    def reject(self, approver: str, reason: str = ""):
        self.approvals[approver] = False
        self.status = "rejected"

    def is_expired(self) -> bool:
        if self.expires_at and datetime.utcnow() > self.expires_at:
            self.status = "expired"
            return True
        return False

class ApprovalEngine:
    def __init__(self):
        self.requests: dict[str, ApprovalRequest] = {}
        self.hooks: dict[str, Callable] = {}

    def create_request(self, type: str, requester: str, data: dict,
                       approvers: list[str], ttl_hours: int = 72,
                       required: int = 1) -> ApprovalRequest:
        req = ApprovalRequest(
            id=str(uuid.uuid4()), type=type, requester=requester,
            data=data, approvers=approvers, required_approvals=required,
            expires_at=datetime.utcnow() + timedelta(hours=ttl_hours),
        )
        self.requests[req.id] = req
        if hook := self.hooks.get(f"on_create_{type}"):
            hook(req)
        return req

    def process_approval(self, request_id: str, approver: str, approved: bool, reason: str = ""):
        req = self.requests[request_id]
        if req.is_expired():
            raise ValueError("Request has expired")
        if approved:
            req.approve(approver)
            if req.status == "approved" and (hook := self.hooks.get(f"on_approve_{req.type}")):
                hook(req)
        else:
            req.reject(approver, reason)
            if hook := self.hooks.get(f"on_reject_{req.type}"):
                hook(req)

Part 4: Saga Pattern with Compensating Transactions

@dataclass
class SagaStep:
    name: str
    action: Callable
    compensate: Callable  # rollback action

class SagaOrchestrator:
    """Execute distributed transactions with compensation on failure."""

    async def execute(self, steps: list[SagaStep], context: dict) -> dict:
        completed: list[SagaStep] = []
        try:
            for step in steps:
                result = await step.action(context)
                context[f"{step.name}_result"] = result
                completed.append(step)
            return context
        except Exception as e:
            # Compensate in reverse order
            for step in reversed(completed):
                try:
                    await step.compensate(context)
                except Exception as comp_error:
                    logger.error("Compensation failed for %s: %s", step.name, comp_error)
            raise

# Example: Transfer money saga
saga = SagaOrchestrator()

transfer_steps = [
    SagaStep("debit_source", debit_account, credit_account_back),
    SagaStep("credit_target", credit_account, debit_account_back),
    SagaStep("record_transfer", save_transfer_record, delete_transfer_record),
    SagaStep("send_notification", notify_users, lambda ctx: None),  # no compensate needed
]

result = await saga.execute(transfer_steps, {"from": "acc1", "to": "acc2", "amount": 100})

Part 5: Task Queue (Custom Async)

import asyncio, json, time, uuid
from dataclasses import dataclass

@dataclass
class Job:
    id: str
    name: str
    payload: dict
    priority: int = 0
    max_retries: int = 3
    attempts: int = 0
    scheduled_at: float | None = None
    status: str = "pending"

class TaskQueue:
    def __init__(self):
        self.queue = asyncio.PriorityQueue()
        self.handlers: dict[str, Callable] = {}
        self.results: dict[str, Any] = {}
        self.dlq: list[Job] = []

    def register(self, name: str, handler: Callable):
        self.handlers[name] = handler

    async def enqueue(self, name: str, payload: dict, priority: int = 0,
                      delay: float = 0) -> str:
        job = Job(
            id=str(uuid.uuid4()), name=name, payload=payload,
            priority=priority,
            scheduled_at=time.time() + delay if delay else None,
        )
        await self.queue.put((priority, time.time(), job))
        return job.id

    async def worker(self, worker_id: str):
        while True:
            _, _, job = await self.queue.get()
            if job.scheduled_at and time.time() < job.scheduled_at:
                await self.queue.put((job.priority, job.scheduled_at, job))
                await asyncio.sleep(1)
                continue
            handler = self.handlers.get(job.name)
            if not handler:
                self.dlq.append(job)
                continue
            try:
                result = await handler(job.payload)
                self.results[job.id] = result
                job.status = "completed"
            except Exception as e:
                job.attempts += 1
                if job.attempts < job.max_retries:
                    await self.queue.put((job.priority + 1, time.time(), job))
                else:
                    job.status = "failed"
                    self.dlq.append(job)

    async def start(self, num_workers: int = 4):
        tasks = [asyncio.create_task(self.worker(f"w{i}")) for i in range(num_workers)]
        return tasks

Part 6: Cron Scheduling

from datetime import datetime
import asyncio

class CronScheduler:
    def __init__(self):
        self.jobs: list[dict] = []

    def schedule(self, name: str, cron_expr: str, handler: Callable):
        self.jobs.append({"name": name, "cron": cron_expr, "handler": handler})

    def _matches(self, cron: str, dt: datetime) -> bool:
        """Simple cron matching: minute hour day month weekday"""
        parts = cron.split()
        checks = [dt.minute, dt.hour, dt.day, dt.month, dt.weekday()]
        for part, value in zip(parts, checks):
            if part == "*":
                continue
            if "/" in part:
                _, step = part.split("/")
                if value % int(step) != 0:
                    return False
            elif "," in part:
                if value not in [int(x) for x in part.split(",")]:
                    return False
            elif int(part) != value:
                return False
        return True

    async def run(self):
        while True:
            now = datetime.utcnow()
            for job in self.jobs:
                if self._matches(job["cron"], now):
                    asyncio.create_task(job["handler"]())
            await asyncio.sleep(60)

scheduler = CronScheduler()
scheduler.schedule("daily_cleanup", "0 3 * * *", cleanup_old_data)
scheduler.schedule("hourly_sync", "0 * * * *", sync_external_data)
scheduler.schedule("every_5_min", "*/5 * * * *", health_check)

AXE MCP Server Integration

Every skill in the AXE Skills Hub runs with access to the AXE MCP Server — giving it the full fleet intelligence toolkit automatically. No setup required; tools are available in any AXE-powered session.

Core Tools Available

Quick Start

# In any AXE session, tools are pre-loaded. Example chaining:

# 1. Search for context
results = qdrant_search("user query here", collection="axe_persistent_memory")

# 2. Fetch live data if needed
content = web_fetch("https://docs.example.com/api")

# 3. Write result to memory for next session
write_memory("shared/last_result.md", output)

# 4. Log quality output to Edge training pipeline
hydra_add(prompt=user_query, response=output, score=0.9, source="skill-name")

Edge Training Integration

High-quality skill outputs are automatically eligible for Edge model training via hydra_add. When a response scores ≥0.85 in evals, pipe it to the Hydra pipeline to compound Edge's knowledge. This is how skills make Edge smarter over time.

# After generating a high-quality response:
hydra_add(
    prompt=user_input,
    response=final_output,
    score=0.9,          # eval score
    source="skill-name" # tracks provenance
)