Tutorial 11 Saga Orchestration¶

# From runtime (convenience re-exports)
from agent_runtime import SagaOrchestrator, SagaState, StepState

# Or directly from hypervisor
from hypervisor.saga.orchestrator import SagaOrchestrator, SagaTimeoutError
from hypervisor.saga.state_machine import Saga, SagaStep, SagaState, StepState, SagaStateError

import asyncio
from hypervisor.saga.orchestrator import SagaOrchestrator
from hypervisor.saga.state_machine import SagaState, StepState


async def main():
    orchestrator = SagaOrchestrator()

    # 1. Create a saga bound to a session
    saga = orchestrator.create_saga(session_id="session-deploy-42")

    # 2. Add steps, each pairing a forward action with a compensation
    step_pr = orchestrator.add_step(
        saga_id=saga.saga_id,
        action_id="data.create_pr",
        agent_did="did:mesh:dev-agent",
        execute_api="/api/pr/create",
        undo_api="/api/pr/close",
        timeout_seconds=60,
        max_retries=2,
    )
    step_tests = orchestrator.add_step(
        saga_id=saga.saga_id,
        action_id="test.run_suite",
        agent_did="did:mesh:ci-agent",
        execute_api="/api/tests/run",
        undo_api="/api/tests/cancel",
        timeout_seconds=300,
    )
    step_deploy = orchestrator.add_step(
        saga_id=saga.saga_id,
        action_id="deploy.staging",
        agent_did="did:mesh:deploy-agent",
        execute_api="/api/deploy/staging",
        undo_api="/api/deploy/rollback",
        timeout_seconds=600,
    )

    # 3. Execute each step with an async callable
    async def create_pr():
        return {"pr_number": 142}

    async def run_tests():
        return {"passed": 247, "failed": 0}

    async def deploy_to_staging():
        raise RuntimeError("Staging cluster unreachable")

    steps_and_executors = [
        (step_pr, create_pr),
        (step_tests, run_tests),
        (step_deploy, deploy_to_staging),
    ]

    for step, executor in steps_and_executors:
        try:
            result = await orchestrator.execute_step(
                saga.saga_id, step.step_id, executor=executor,
            )
            print(f"  ✓ {step.action_id} committed: {result}")
        except Exception as e:
            print(f"  ✗ {step.action_id} failed: {e}")
            break

    # 4. Compensate all committed steps in reverse order
    async def compensator(step):
        print(f"  ↩ Compensating {step.action_id} via {step.undo_api}")
        return "compensated"

    failed = await orchestrator.compensate(saga.saga_id, compensator)
    print(f"Saga state: {saga.state}")
    # SagaState.COMPLETED (all compensations succeeded)


asyncio.run(main())

class SagaOrchestrator:
    DEFAULT_MAX_RETRIES = 2
    DEFAULT_RETRY_DELAY_SECONDS = 1.0

    def create_saga(self, session_id: str) -> Saga
    def add_step(self, saga_id, action_id, agent_did, execute_api,
                 undo_api=None, timeout_seconds=300, max_retries=0) -> SagaStep
    async def execute_step(self, saga_id, step_id, executor: Callable) -> Any
    async def compensate(self, saga_id, compensator: Callable) -> list[SagaStep]
    def get_saga(self, saga_id: str) -> Saga | None
    active_sagas: list[Saga]  # property

async def fetch_data():
    response = await http_client.get("https://api.example.com/data")
    return response.json()

result = await orchestrator.execute_step(
    saga.saga_id,
    step.step_id,
    executor=fetch_data,
)
# On success: step.state == StepState.COMMITTED
# On failure: step.state == StepState.FAILED, raises the exception

# Get all sagas that haven't reached a terminal state
active = orchestrator.active_sagas

# Look up a specific saga by ID
saga = orchestrator.get_saga("saga:a1b2c3d4-...")
if saga:
    print(f"State: {saga.state}, Steps: {len(saga.steps)}")

from hypervisor.saga.state_machine import SagaStep, StepState, SagaStateError

step = SagaStep(
    step_id="s1",
    action_id="data.extract",
    agent_did="did:mesh:etl-agent",
    execute_api="/api/extract",
)

# Valid transitions
step.transition(StepState.EXECUTING)    # PENDING → EXECUTING ✓
assert step.started_at is not None      # timestamp set automatically

step.transition(StepState.COMMITTED)    # EXECUTING → COMMITTED ✓
assert step.completed_at is not None

# Invalid transition raises SagaStateError
try:
    step.transition(StepState.PENDING)  # COMMITTED → PENDING ✗
except SagaStateError as e:
    print(e)  # "Invalid step transition: committed → pending"

from hypervisor.saga.state_machine import Saga, SagaState

saga = Saga(saga_id="saga:1", session_id="session:1")
assert saga.state == SagaState.RUNNING

# Saga transitions are also validated
saga.transition(SagaState.COMPENSATING)  # RUNNING → COMPENSATING ✓
saga.transition(SagaState.COMPLETED)     # COMPENSATING → COMPLETED ✓
assert saga.completed_at is not None

# Serialize saga to a dictionary
saga_dict = saga.to_dict()
# {"saga_id": "saga:...", "session_id": "...", "state": "running", "steps": [...]}

# Inspect committed steps (execution order and reverse/compensation order)
for step in saga.committed_steps:
    print(f"{step.action_id}: {step.execute_result}")

for step in saga.committed_steps_reversed:
    print(f"Would compensate: {step.action_id}")

async def compensator(step: SagaStep) -> Any:
    """Called for each committed step in reverse order."""
    print(f"Undoing {step.action_id} via {step.undo_api}")
    return "compensated"

failed_steps = await orchestrator.compensate(saga.saga_id, compensator)

async def failing_compensator(step):
    raise RuntimeError("Cannot rollback")

failed = await orchestrator.compensate(saga.saga_id, failing_compensator)
assert saga.state == SagaState.ESCALATED
assert len(failed) > 0
assert failed[0].state == StepState.COMPENSATION_FAILED

from hypervisor.saga.state_machine import SagaStateError
from hypervisor.saga.orchestrator import SagaTimeoutError

step = orchestrator.add_step(
    saga_id=saga.saga_id,
    action_id="data.long_process",
    agent_did="did:mesh:processor",
    execute_api="/api/process",
    timeout_seconds=10,
)

async def slow_executor():
    await asyncio.sleep(30)  # Exceeds timeout
    return "done"

try:
    await orchestrator.execute_step(saga.saga_id, step.step_id, executor=slow_executor)
except SagaTimeoutError:
    print(f"Step state: {step.state}")  # StepState.FAILED

attempt_count = 0

async def flaky_executor():
    global attempt_count
    attempt_count += 1
    if attempt_count < 3:
        raise ConnectionError("Temporarily unavailable")
    return "success on attempt 3"

step = orchestrator.add_step(
    saga_id=saga.saga_id,
    action_id="data.fetch",
    agent_did="did:mesh:fetcher",
    execute_api="/api/fetch",
    max_retries=2,  # 1 initial + 2 retries = 3 total attempts
)

result = await orchestrator.execute_step(
    saga.saga_id, step.step_id, executor=flaky_executor,
)
assert step.state == StepState.COMMITTED
assert step.retry_count == 2

async def run_saga_safely(orchestrator, saga, steps_and_executors, compensator):
    """Execute a saga with automatic compensation on failure."""
    for step, executor in steps_and_executors:
        try:
            await orchestrator.execute_step(
                saga.saga_id, step.step_id, executor=executor,
            )
        except Exception:
            failed_compensations = await orchestrator.compensate(
                saga.saga_id, compensator,
            )
            if saga.state == SagaState.ESCALATED:
                raise RuntimeError(
                    f"Saga ESCALATED: {len(failed_compensations)} "
                    "compensation(s) failed. Human intervention required."
                )
            return {"status": "rolled_back", "failed_at": step.action_id}

    return {"status": "committed", "steps": len(steps_and_executors)}

from hypervisor import ExecutionRing, ReversibilityLevel
from hypervisor.models import ActionDescriptor
from hypervisor.rings.classifier import ActionClassifier
from hypervisor.saga.orchestrator import SagaOrchestrator

classifier = ActionClassifier()
orchestrator = SagaOrchestrator()

saga = orchestrator.create_saga("session-governed-deploy")
step = orchestrator.add_step(
    saga_id=saga.saga_id,
    action_id="deploy.production",
    agent_did="did:mesh:deploy-bot",
    execute_api="/api/deploy/prod",
    undo_api="/api/deploy/rollback",
)

# Classify the action, then compare its required ring to the agent ring
action = ActionDescriptor(
    action_id="deploy.production",
    name="Deploy to production",
    execute_api="/api/deploy/prod",
    undo_api="/api/deploy/rollback",
    reversibility=ReversibilityLevel.PARTIAL,
)
classification = classifier.classify(action)
agent_ring = ExecutionRing.from_eff_score(eff_score=0.72)

if classification.ring.value < agent_ring.value:
    print(f"Agent ring {agent_ring} insufficient for {classification.ring}")
    await orchestrator.compensate(saga.saga_id, compensator)
else:
    await orchestrator.execute_step(saga.saga_id, step.step_id, executor=deploy_fn)

import asyncio
from hypervisor.saga.orchestrator import SagaOrchestrator
from hypervisor.saga.state_machine import SagaState

# ── 1. Create the orchestrator and a saga for the pipeline ───────

orchestrator = SagaOrchestrator()
saga = orchestrator.create_saga(session_id="pipeline-2025-w03")

# ── 2. Add the pipeline steps in execution order ─────────────────

pipeline = [
    # action_id, agent_did, execute_api, undo_api, timeout, retries
    ("data.extract", "did:mesh:extractor", "/api/extract/sales",
     "/api/extract/cleanup", 120, 2),
    ("data.transform", "did:mesh:transformer", "/api/transform",
     "/api/transform/rollback", 600, 0),
    ("validate.quality", "did:mesh:validator", "/api/validate",
     "/api/validate/reset", 300, 0),
    ("data.load", "did:mesh:loader", "/api/load/warehouse",
     "/api/load/rollback", 900, 0),
    ("notify.team", "did:mesh:notifier", "/api/notify/slack",
     None, 60, 0),
]

steps = [
    orchestrator.add_step(
        saga_id=saga.saga_id,
        action_id=action_id,
        agent_did=agent_did,
        execute_api=execute_api,
        undo_api=undo_api,
        timeout_seconds=timeout,
        max_retries=retries,
    )
    for action_id, agent_did, execute_api, undo_api, timeout, retries in pipeline
]

# ── 3. Provide an async executor for each step ───────────────────

async def extract():   return {"records": 23_720}
async def transform(): return {"records": 23_720}
async def validate():  return {"score": 0.97}
async def load():      return {"rows_inserted": 23_720}
async def notify():    return {"sent": True}

executors = [extract, transform, validate, load, notify]

# ── 4. Compensator called for each committed step on rollback ────

async def compensator(step):
    print(f"  ↩ Compensating {step.action_id} via {step.undo_api}")
    return "compensated"

# ── 5. Run the pipeline, compensating on the first failure ───────

async def run_pipeline():
    for step, executor in zip(steps, executors):
        try:
            result = await orchestrator.execute_step(
                saga.saga_id, step.step_id, executor=executor,
            )
            print(f"  ✓ {step.action_id}: {result}")
        except Exception as e:
            print(f"  ✗ {step.action_id} failed: {e}")
            failed = await orchestrator.compensate(saga.saga_id, compensator)
            if saga.state == SagaState.ESCALATED:
                raise RuntimeError(
                    f"{len(failed)} compensation(s) failed; manual repair required"
                )
            return

    print(f"Pipeline complete, saga state: {saga.state}")

asyncio.run(run_pipeline())