saga-orchestration

name	saga-orchestration
description	Patterns for managing distributed transactions and long-running business processes.
category	AI & Agents
source	antigravity
tags	["python","ai","llm","automation","workflow","template","design","cro"]
url	https://github.com/sickn33/antigravity-awesome-skills/tree/main/skills/saga-orchestration

Saga Orchestration

Patterns for managing distributed transactions and long-running business processes.

Do not use this skill when

• The task is unrelated to saga orchestration
• You need a different domain or tool outside this scope

Instructions

• Clarify goals, constraints, and required inputs.
• Apply relevant best practices and validate outcomes.
• Provide actionable steps and verification.
• If detailed examples are required, open resources/implementation-playbook.md.

Use this skill when

• Coordinating multi-service transactions
• Implementing compensating transactions
• Managing long-running business workflows
• Handling failures in distributed systems
• Building order fulfillment processes
• Implementing approval workflows

Core Concepts

1. Saga Types

Choreography                    Orchestration
┌─────┐  ┌─────┐  ┌─────┐     ┌─────────────┐
│Svc A│─►│Svc B│─►│Svc C│     │ Orchestrator│
└─────┘  └─────┘  └─────┘     └──────┬──────┘
   │        │        │               │
   ▼        ▼        ▼         ┌─────┼─────┐
 Event    Event    Event       ▼     ▼     ▼
                            ┌────┐┌────┐┌────┐
                            │Svc1││Svc2││Svc3│
                            └────┘└────┘└────┘

2. Saga Execution States

State	Description
Started	Saga initiated
Pending	Waiting for step completion
Compensating	Rolling back due to failure
Completed	All steps succeeded
Failed	Saga failed after compensation

Templates

Template 1: Saga Orchestrator Base

from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from enum import Enum
from typing import List, Dict, Any, Optional
from datetime import datetime
import uuid

class SagaState(Enum):
    STARTED = "started"
    PENDING = "pending"
    COMPENSATING = "compensating"
    COMPLETED = "completed"
    FAILED = "failed"


@dataclass
class SagaStep:
    name: str
    action: str
    compensation: str
    status: str = "pending"
    result: Optional[Dict] = None
    error: Optional[str] = None
    executed_at: Optional[datetime] = None
    compensated_at: Optional[datetime] = None


@dataclass
class Saga:
    saga_id: str
    saga_type: str
    state: SagaState
    data: Dict[str, Any]
    steps: List[SagaStep]
    current_step:  = 
    created_at: datetime = field(default_factory=datetime.utcnow)
    updated_at: datetime = field(default_factory=datetime.utcnow)


 ():
    

     ():
        .saga_store = saga_store
        .event_publisher = event_publisher


     () -> [SagaStep]:
        
        



     () -> :
        
        

      () -> Saga:
        
        saga = Saga(
            saga_id=(uuid.uuid4()),
            saga_type=.saga_type,
            state=SagaState.STARTED,
            data=data,
            steps=.define_steps(data)
        )
         .saga_store.save(saga)
         ._execute_next_step(saga)
         saga

      ():
        
        saga =  .saga_store.get(saga_id)

        
         step  saga.steps:
             step.name == step_name:
                step.status = 
                step.result = result
                step.executed_at = datetime.utcnow()
                

        saga.current_step += 
        saga.updated_at = datetime.utcnow()

        
         saga.current_step >= (saga.steps):
            saga.state = SagaState.COMPLETED
             .saga_store.save(saga)
             ._on_saga_completed(saga)
        :
            saga.state = SagaState.PENDING
             .saga_store.save(saga)
             ._execute_next_step(saga)

      ():
        
        saga =  .saga_store.get(saga_id)

        
         step  saga.steps:
             step.name == step_name:
                step.status = 
                step.error = error
                

        saga.state = SagaState.COMPENSATING
        saga.updated_at = datetime.utcnow()
         .saga_store.save(saga)

        
         ._compensate(saga)