Saga Orchestration Deep Dive.

# Normal code - NOT durable

async def process_order(order_id: str):
    # Step 1: Reserve inventory
    reservation = await inventory_service.reserve(order_id)
    
    # CRASH HERE - what happens?
    # - reservation exists in inventory service
    # - but we lost track of it
    # - if we restart, do we reserve again? (double reserve!)
    # - if we don't, order is stuck
    
    # Step 2: Charge payment
    payment = await payment_service.charge(order_id)
    
    # Step 3: Ship
    shipment = await shipping_service.create(order_id)
    
    return shipment.tracking_number

# Durable execution with Temporal

@workflow.defn
class OrderWorkflow:
    @workflow.run
    async def run(self, order_id: str) -> str:
        # Step 1: Reserve inventory
        reservation = await workflow.execute_activity(
            reserve_inventory,
            order_id,
            start_to_close_timeout=timedelta(minutes=5)
        )
        
        # CRASH HERE - no problem!
        # Temporal recorded that reserve_inventory completed
        # On restart, it SKIPS step 1 and continues from step 2
        # The reservation result is replayed from history
        
        # Step 2: Charge payment
        payment = await workflow.execute_activity(
            charge_payment,
            order_id,
            start_to_close_timeout=timedelta(minutes=5)
        )
        
        # Step 3: Ship
        shipment = await workflow.execute_activity(
            create_shipment,
            order_id,
            start_to_close_timeout=timedelta(minutes=5)
        )
        
        return shipment["tracking_number"]

# Basic Temporal Workflow

from datetime import timedelta
from temporalio import workflow, activity
from temporalio.common import RetryPolicy
from dataclasses import dataclass
from typing import Optional


# =============================================================================
# Data Classes
# =============================================================================

@dataclass
class OrderInput:
    order_id: str
    customer_id: str
    items: list
    total_amount: float
    shipping_address: str


@dataclass
class OrderResult:
    order_id: str
    status: str
    tracking_number: Optional[str] = None
    error: Optional[str] = None


# =============================================================================
# Activities (Side Effects)
# =============================================================================

@activity.defn
async def reserve_inventory(order_id: str, items: list) -> dict:
    """
    Activity: Reserve inventory.
    
    This is where actual work happens (DB calls, API calls).
    Activities can fail and will be retried according to policy.
    """
    async with get_db_connection() as db, db.transaction():
        # Activities may be retried: one reservation per order, all-or-nothing
        existing = await db.fetch_val(
            "SELECT id FROM reservations WHERE order_id = $1", order_id
        )
        if existing:
            return {"reservation_id": str(existing)}
        
        reservation_id = await db.fetch_val(
            "INSERT INTO reservations (order_id, items, status) VALUES ($1, $2, 'active') RETURNING id",
            order_id, json.dumps(items)
        )
        
        for item in items:
            result = await db.execute(
                "UPDATE inventory SET quantity = quantity - $1 WHERE product_id = $2 AND quantity >= $1",
                item['quantity'], item['product_id']
            )
            if result.rowcount == 0:
                raise InsufficientInventoryError(f"Not enough {item['product_id']}")
        
        return {"reservation_id": str(reservation_id)}


@activity.defn
async def release_inventory(reservation_id: str) -> dict:
    """Activity: Release inventory reservation (compensation)."""
    async with get_db_connection() as db, db.transaction():
        reservation = await db.fetch_one(
            "SELECT items FROM reservations WHERE id = $1 AND status = 'active' FOR UPDATE",
            reservation_id
        )
        
        if not reservation:
            return {"status": "already_released"}
        
        items = json.loads(reservation['items'])
        for item in items:
            await db.execute(
                "UPDATE inventory SET quantity = quantity + $1 WHERE product_id = $2",
                item['quantity'], item['product_id']
            )
        
        await db.execute(
            "UPDATE reservations SET status = 'released' WHERE id = $1",
            reservation_id
        )
        
        return {"status": "released"}


@activity.defn
async def charge_payment(
    customer_id: str,
    amount: float,
    idempotency_key: str
) -> dict:
    """Activity: Charge customer payment."""
    result = await stripe.charges.create(
        amount=int(amount * 100),
        currency="usd",
        customer=customer_id,
        idempotency_key=idempotency_key
    )
    return {"payment_id": result.id, "status": result.status}


@activity.defn
async def refund_payment(payment_id: str, idempotency_key: str) -> dict:
    """Activity: Refund payment (compensation)."""
    result = await stripe.refunds.create(
        charge=payment_id,  # payment_id is the Charge id from charge_payment
        idempotency_key=idempotency_key
    )
    return {"refund_id": result.id, "status": result.status}


@activity.defn
async def create_shipment(order_id: str, address: str, items: list) -> dict:
    """Activity: Create shipment with carrier."""
    shipment = await shipping_api.create_shipment(
        reference=order_id,
        destination=address,
        items=items
    )
    return {
        "shipment_id": shipment.id,
        "tracking_number": shipment.tracking_number
    }


@activity.defn
async def send_notification(
    customer_id: str,
    notification_type: str,
    data: dict
) -> dict:
    """Activity: Send notification to customer."""
    await notification_service.send(
        customer_id=customer_id,
        type=notification_type,
        data=data
    )
    return {"sent": True}


# =============================================================================
# Workflow (Orchestration Logic)
# =============================================================================

@workflow.defn
class OrderWorkflow:
    """
    Order processing workflow with saga compensation.
    
    This workflow:
    1. Reserves inventory
    2. Charges payment
    3. Creates shipment (pivot point)
    4. Sends confirmation
    
    If any step before shipment fails, it compensates.
    """
    
    def __init__(self):
        self.status = "pending"
        self.reservation_id = None
        self.payment_id = None
        self.tracking_number = None
    
    @workflow.run
    async def run(self, input: OrderInput) -> OrderResult:
        """Main workflow execution."""
        
        # Retry policy for activities
        retry_policy = RetryPolicy(
            initial_interval=timedelta(seconds=1),
            maximum_interval=timedelta(minutes=1),
            backoff_coefficient=2.0,
            maximum_attempts=5,
            non_retryable_error_types=["InsufficientInventoryError", "PaymentDeclinedError"]
        )
        
        try:
            # Step 1: Reserve inventory
            self.status = "reserving_inventory"
            reservation = await workflow.execute_activity(
                reserve_inventory,
                args=[input.order_id, input.items],
                start_to_close_timeout=timedelta(minutes=5),
                retry_policy=retry_policy
            )
            self.reservation_id = reservation["reservation_id"]
            
            # Step 2: Charge payment
            self.status = "charging_payment"
            payment = await workflow.execute_activity(
                charge_payment,
                args=[input.customer_id, input.total_amount, f"order-{input.order_id}"],
                start_to_close_timeout=timedelta(minutes=5),
                retry_policy=retry_policy
            )
            self.payment_id = payment["payment_id"]
            
            # Step 3: Create shipment (PIVOT - no compensation after this)
            self.status = "creating_shipment"
            shipment = await workflow.execute_activity(
                create_shipment,
                args=[input.order_id, input.shipping_address, input.items],
                start_to_close_timeout=timedelta(minutes=10),
                retry_policy=RetryPolicy(maximum_attempts=0)  # Retry forever after pivot
            )
            self.tracking_number = shipment["tracking_number"]
            
            # Step 4: Send confirmation
            self.status = "sending_confirmation"
            await workflow.execute_activity(
                send_notification,
                args=[input.customer_id, "order_confirmation", {
                    "order_id": input.order_id,
                    "tracking_number": self.tracking_number
                }],
                start_to_close_timeout=timedelta(minutes=1),
                retry_policy=RetryPolicy(maximum_attempts=0)  # After pivot: retry, never compensate
            )
            
            self.status = "completed"
            return OrderResult(
                order_id=input.order_id,
                status="completed",
                tracking_number=self.tracking_number
            )
            
        except Exception as e:
            # Compensation logic
            workflow.logger.error(f"Order workflow failed: {e}")
            await self._compensate(input)
            
            return OrderResult(
                order_id=input.order_id,
                status="failed",
                error=str(e)
            )
    
    async def _compensate(self, input: OrderInput):
        """Execute compensation for completed steps."""
        self.status = "compensating"
        
        compensation_policy = RetryPolicy(
            maximum_attempts=0  # Retry forever - compensation must succeed
        )
        
        # Refund payment if charged
        if self.payment_id:
            try:
                await workflow.execute_activity(
                    refund_payment,
                    args=[self.payment_id, f"refund-{input.order_id}"],
                    start_to_close_timeout=timedelta(minutes=5),
                    retry_policy=compensation_policy
                )
            except Exception as e:
                workflow.logger.error(f"Refund failed: {e}")
        
        # Release inventory if reserved
        if self.reservation_id:
            try:
                await workflow.execute_activity(
                    release_inventory,
                    args=[self.reservation_id],
                    start_to_close_timeout=timedelta(minutes=5),
                    retry_policy=compensation_policy
                )
            except Exception as e:
                workflow.logger.error(f"Release inventory failed: {e}")
        
        # Notify customer of failure
        await workflow.execute_activity(
            send_notification,
            args=[input.customer_id, "order_failed", {
                "order_id": input.order_id,
                "reason": "Payment or inventory issue"
            }],
            start_to_close_timeout=timedelta(minutes=1),
            retry_policy=compensation_policy
        )
        
        self.status = "compensated"
    
    @workflow.query
    def get_status(self) -> str:
        """Query: Get current workflow status."""
        return self.status
    
    @workflow.query
    def get_tracking_number(self) -> Optional[str]:
        """Query: Get tracking number if available."""
        return self.tracking_number

# Workflows with Signals

@workflow.defn
class OrderWithApprovalWorkflow:
    """
    Order workflow that waits for manager approval.
    
    Demonstrates:
    - Signals for external events
    - Waiting for signals with timeout
    - Workflow state queries
    """
    
    def __init__(self):
        self.status = "pending"
        self.approved = None  # None = waiting, True/False = decided
        self.approver = None
    
    @workflow.run
    async def run(self, input: OrderInput) -> OrderResult:
        # Step 1: Reserve inventory
        reservation = await workflow.execute_activity(
            reserve_inventory,
            args=[input.order_id, input.items],
            start_to_close_timeout=timedelta(minutes=5)
        )
        
        # Step 2: Wait for approval if order > $1000
        if input.total_amount > 1000:
            self.status = "pending_approval"
            
            # Send approval request notification
            await workflow.execute_activity(
                send_notification,
                args=["managers", "approval_required", {
                    "order_id": input.order_id,
                    "amount": input.total_amount,
                    "customer": input.customer_id
                }],
                start_to_close_timeout=timedelta(minutes=1)
            )
            
            # Wait for approval signal (up to 24 hours)
            try:
                await workflow.wait_condition(
                    lambda: self.approved is not None,
                    timeout=timedelta(hours=24)
                )
            except asyncio.TimeoutError:
                # No approval received - auto-reject
                self.approved = False
                self.approver = "system_timeout"
            
            if not self.approved:
                # Rejected - compensate and return
                await self._compensate(input, reservation["reservation_id"])
                return OrderResult(
                    order_id=input.order_id,
                    status="rejected",
                    error=f"Rejected by {self.approver}"
                )
        
        # Step 3: Continue with payment and shipment
        self.status = "processing"
        # ... rest of workflow
    
    @workflow.signal
    async def approve(self, approver: str):
        """Signal: Approve the order."""
        self.approved = True
        self.approver = approver
    
    @workflow.signal
    async def reject(self, approver: str, reason: str):
        """Signal: Reject the order."""
        self.approved = False
        self.approver = approver
        self.rejection_reason = reason
    
    @workflow.query
    def get_approval_status(self) -> dict:
        """Query: Get approval status."""
        return {
            "status": self.status,
            "approved": self.approved,
            "approver": self.approver
        }


# Client code to send signals
async def approve_order(order_id: str, manager_id: str):
    """Manager approves an order."""
    client = await Client.connect("localhost:7233")
    
    handle = client.get_workflow_handle(f"order-{order_id}")
    await handle.signal(OrderWithApprovalWorkflow.approve, manager_id)


async def reject_order(order_id: str, manager_id: str, reason: str):
    """Manager rejects an order."""
    client = await Client.connect("localhost:7233")
    
    handle = client.get_workflow_handle(f"order-{order_id}")
    await handle.signal(OrderWithApprovalWorkflow.reject, args=[manager_id, reason])

# Parent-Child Workflow Pattern

@workflow.defn
class BatchOrderWorkflow:
    """
    Process multiple orders as a batch.
    
    Each order runs as a child workflow.
    Parent tracks overall progress.
    """
    
    @workflow.run
    async def run(self, order_ids: list[str]) -> dict:
        results = {}
        
        # Start all orders as child workflows
        handles = []
        for order_id in order_ids:
            # Get order details (this would be an activity in real code)
            order_input = OrderInput(
                order_id=order_id,
                customer_id="batch-customer",
                items=[],
                total_amount=100.0,
                shipping_address="123 Main St"
            )
            
            # Start child workflow
            handle = await workflow.start_child_workflow(
                OrderWorkflow.run,
                order_input,
                id=f"order-{order_id}",
                task_queue="orders"
            )
            handles.append((order_id, handle))
        
        # Wait for all child workflows to complete
        for order_id, handle in handles:
            try:
                result = await handle
                results[order_id] = {
                    "status": result.status,
                    "tracking": result.tracking_number
                }
            except Exception as e:
                results[order_id] = {
                    "status": "failed",
                    "error": str(e)
                }
        
        return {
            "total": len(order_ids),
            "succeeded": sum(1 for r in results.values() if r["status"] == "completed"),
            "failed": sum(1 for r in results.values() if r["status"] == "failed"),
            "results": results
        }

# State Machine Pattern for Complex Sagas

from enum import Enum
from typing import Dict, Callable, Optional
from dataclasses import dataclass


class SagaState(Enum):
    INITIALIZED = "initialized"
    RESERVING_INVENTORY = "reserving_inventory"
    INVENTORY_RESERVED = "inventory_reserved"
    CHARGING_PAYMENT = "charging_payment"
    PAYMENT_CHARGED = "payment_charged"
    CREATING_SHIPMENT = "creating_shipment"
    SHIPMENT_CREATED = "shipment_created"
    COMPLETING = "completing"
    COMPLETED = "completed"
    COMPENSATING = "compensating"
    COMPENSATED = "compensated"
    FAILED = "failed"


@dataclass
class StateTransition:
    from_state: SagaState
    to_state: SagaState
    action: Optional[Callable] = None
    compensation: Optional[Callable] = None


@workflow.defn
class StateMachineOrderWorkflow:
    """
    Order workflow using explicit state machine.
    
    Benefits:
    - Clear visualization of all states
    - Explicit transitions
    - Easy to add new states
    - Better for complex branching
    """
    
    # State machine definition
    TRANSITIONS = {
        SagaState.INITIALIZED: [
            StateTransition(
                SagaState.INITIALIZED,
                SagaState.RESERVING_INVENTORY
            )
        ],
        SagaState.RESERVING_INVENTORY: [
            StateTransition(
                SagaState.RESERVING_INVENTORY,
                SagaState.INVENTORY_RESERVED,
                action="reserve_inventory",
                compensation="release_inventory"
            ),
            StateTransition(
                SagaState.RESERVING_INVENTORY,
                SagaState.FAILED  # On error
            )
        ],
        SagaState.INVENTORY_RESERVED: [
            StateTransition(
                SagaState.INVENTORY_RESERVED,
                SagaState.CHARGING_PAYMENT
            )
        ],
        SagaState.CHARGING_PAYMENT: [
            StateTransition(
                SagaState.CHARGING_PAYMENT,
                SagaState.PAYMENT_CHARGED,
                action="charge_payment",
                compensation="refund_payment"
            ),
            StateTransition(
                SagaState.CHARGING_PAYMENT,
                SagaState.COMPENSATING  # On error, start compensation
            )
        ],
        # ... more transitions
    }
    
    def __init__(self):
        self.state = SagaState.INITIALIZED
        self.history: list[SagaState] = []
        self.results: dict = {}
    
    @workflow.run
    async def run(self, input: OrderInput) -> OrderResult:
        """Execute state machine."""
        
        while self.state not in [SagaState.COMPLETED, SagaState.COMPENSATED, SagaState.FAILED]:
            next_state = await self._execute_state(input)
            self.history.append(self.state)
            self.state = next_state
            
            workflow.logger.info(f"State transition: {self.history[-1]} -> {self.state}")
        
        return OrderResult(
            order_id=input.order_id,
            status=self.state.value,
            tracking_number=self.results.get("tracking_number")
        )
    
    async def _execute_state(self, input: OrderInput) -> SagaState:
        """Execute current state and return next state."""
        
        if self.state == SagaState.INITIALIZED:
            return SagaState.RESERVING_INVENTORY
        
        elif self.state == SagaState.RESERVING_INVENTORY:
            try:
                result = await workflow.execute_activity(
                    reserve_inventory,
                    args=[input.order_id, input.items],
                    start_to_close_timeout=timedelta(minutes=5)
                )
                self.results["reservation_id"] = result["reservation_id"]
                return SagaState.INVENTORY_RESERVED
            except Exception:
                return SagaState.FAILED
        
        elif self.state == SagaState.INVENTORY_RESERVED:
            return SagaState.CHARGING_PAYMENT
        
        elif self.state == SagaState.CHARGING_PAYMENT:
            try:
                result = await workflow.execute_activity(
                    charge_payment,
                    args=[input.customer_id, input.total_amount, f"order-{input.order_id}"],
                    start_to_close_timeout=timedelta(minutes=5)
                )
                self.results["payment_id"] = result["payment_id"]
                return SagaState.PAYMENT_CHARGED
            except Exception:
                return SagaState.COMPENSATING
        
        elif self.state == SagaState.PAYMENT_CHARGED:
            return SagaState.CREATING_SHIPMENT
        
        elif self.state == SagaState.CREATING_SHIPMENT:
            # This is the pivot - retry forever
            result = await workflow.execute_activity(
                create_shipment,
                args=[input.order_id, input.shipping_address, input.items],
                start_to_close_timeout=timedelta(minutes=10),
                retry_policy=RetryPolicy(maximum_attempts=0)
            )
            self.results["tracking_number"] = result["tracking_number"]
            return SagaState.SHIPMENT_CREATED
        
        elif self.state == SagaState.SHIPMENT_CREATED:
            return SagaState.COMPLETING
        
        elif self.state == SagaState.COMPLETING:
            await workflow.execute_activity(
                send_notification,
                args=[input.customer_id, "order_confirmation", self.results],
                start_to_close_timeout=timedelta(minutes=1)
            )
            return SagaState.COMPLETED
        
        elif self.state == SagaState.COMPENSATING:
            await self._run_compensation(input)
            return SagaState.COMPENSATED
        
        return SagaState.FAILED
    
    async def _run_compensation(self, input: OrderInput):
        """Run compensation based on history."""
        # Compensate in reverse order
        for state in reversed(self.history):
            if state == SagaState.CHARGING_PAYMENT and "payment_id" in self.results:
                await workflow.execute_activity(
                    refund_payment,
                    args=[self.results["payment_id"], f"refund-{input.order_id}"],
                    start_to_close_timeout=timedelta(minutes=5),
                    retry_policy=RetryPolicy(maximum_attempts=0)
                )
            
            elif state == SagaState.RESERVING_INVENTORY and "reservation_id" in self.results:
                await workflow.execute_activity(
                    release_inventory,
                    args=[self.results["reservation_id"]],
                    start_to_close_timeout=timedelta(minutes=5),
                    retry_policy=RetryPolicy(maximum_attempts=0)
                )
    
    @workflow.query
    def get_state(self) -> dict:
        """Query current state machine status."""
        return {
            "current_state": self.state.value,
            "history": [s.value for s in self.history],
            "results": self.results
        }

# Workflow Versioning for Safe Deployments

@workflow.defn
class VersionedOrderWorkflow:
    """
    Workflow with versioning for safe deployments.
    
    When you need to change workflow logic, use versioning
    to ensure in-flight workflows use old code while
    new workflows use new code.
    """
    
    @workflow.run
    async def run(self, input: OrderInput) -> OrderResult:
        # Reserve inventory (unchanged)
        reservation = await workflow.execute_activity(
            reserve_inventory,
            args=[input.order_id, input.items],
            start_to_close_timeout=timedelta(minutes=5)
        )
        
        # Version 1: Original flow - charge then ship
        # Version 2: New flow - fraud check before charge
        
        version = workflow.patched("fraud-check-v2")
        
        if version:
            # New code path: Add fraud check
            fraud_result = await workflow.execute_activity(
                check_fraud,
                args=[input.customer_id, input.total_amount],
                start_to_close_timeout=timedelta(minutes=2)
            )
            
            if fraud_result["risk_score"] > 0.8:
                # High risk - require manual review
                await workflow.wait_condition(
                    lambda: self.fraud_approved is not None,
                    timeout=timedelta(hours=24)
                )
                
                if not self.fraud_approved:
                    await self._compensate(input, reservation)
                    return OrderResult(
                        order_id=input.order_id,
                        status="rejected_fraud"
                    )
        
        # Continue with payment (both versions)
        payment = await workflow.execute_activity(
            charge_payment,
            args=[input.customer_id, input.total_amount, f"order-{input.order_id}"],
            start_to_close_timeout=timedelta(minutes=5)
        )
        
        # Version 3: Add loyalty points
        version_loyalty = workflow.patched("loyalty-points-v3")
        
        if version_loyalty:
            await workflow.execute_activity(
                award_loyalty_points,
                args=[input.customer_id, input.total_amount],
                start_to_close_timeout=timedelta(minutes=1)
            )
        
        # Continue with shipment...

# Long-Running Subscription Workflow

@workflow.defn
class SubscriptionWorkflow:
    """
    Subscription lifecycle workflow.
    
    This workflow can run for YEARS:
    - Handles monthly billing
    - Processes upgrades/downgrades
    - Manages cancellation
    """
    
    def __init__(self):
        self.status = "active"
        self.current_plan = None
        self.next_billing_date = None
        self.cancelled = False
        self.upgrade_request = None
    
    @workflow.run
    async def run(self, customer_id: str, initial_plan: str) -> dict:
        """Main subscription loop."""
        self.current_plan = initial_plan
        self.next_billing_date = workflow.now() + timedelta(days=30)
        
        while not self.cancelled:
            # Wait until next billing date OR a signal arrives
            try:
                await workflow.wait_condition(
                    lambda: self.cancelled or self.upgrade_request is not None,
                    timeout=self._time_until_billing()
                )
            except asyncio.TimeoutError:
                # Billing time!
                pass
            
            # Check for cancellation
            if self.cancelled:
                break
            
            # Check for upgrade/downgrade
            if self.upgrade_request:
                await self._process_plan_change()
                self.upgrade_request = None
                continue
            
            # Process billing
            await self._process_billing(customer_id)
            self.next_billing_date = workflow.now() + timedelta(days=30)
        
        # Subscription ended
        await self._process_cancellation(customer_id)
        
        return {
            "status": "cancelled",
            "final_billing_date": str(self.next_billing_date)
        }
    
    async def _process_billing(self, customer_id: str):
        """Process monthly billing."""
        plan_details = await workflow.execute_activity(
            get_plan_details,
            args=[self.current_plan],
            start_to_close_timeout=timedelta(minutes=1)
        )
        
        try:
            await workflow.execute_activity(
                charge_subscription,
                args=[customer_id, plan_details["price"], f"sub-{workflow.info().workflow_id}-{workflow.now().isoformat()}"],
                start_to_close_timeout=timedelta(minutes=5),
                retry_policy=RetryPolicy(
                    maximum_attempts=5,
                    initial_interval=timedelta(hours=1)  # Waits 1+2+4+8h: retries span ~15 hours
                )
            )
            self.status = "active"
            
        except Exception as e:
            self.status = "payment_failed"
            # Could implement grace period, dunning, etc.
    
    async def _process_plan_change(self):
        """Process plan upgrade/downgrade."""
        new_plan = self.upgrade_request
        
        # Prorate current billing period
        await workflow.execute_activity(
            prorate_subscription,
            args=[self.current_plan, new_plan, self.next_billing_date],
            start_to_close_timeout=timedelta(minutes=5)
        )
        
        self.current_plan = new_plan
    
    async def _process_cancellation(self, customer_id: str):
        """Process subscription cancellation."""
        await workflow.execute_activity(
            send_notification,
            args=[customer_id, "subscription_cancelled", {
                "plan": self.current_plan
            }],
            start_to_close_timeout=timedelta(minutes=1)
        )
    
    def _time_until_billing(self) -> timedelta:
        """Calculate time until next billing."""
        now = workflow.now()
        if self.next_billing_date > now:
            return self.next_billing_date - now
        return timedelta(seconds=0)
    
    @workflow.signal
    async def cancel(self):
        """Signal: Cancel subscription."""
        self.cancelled = True
    
    @workflow.signal
    async def change_plan(self, new_plan: str):
        """Signal: Change subscription plan."""
        self.upgrade_request = new_plan
    
    @workflow.query
    def get_subscription_status(self) -> dict:
        """Query: Get current subscription status."""
        return {
            "status": self.status,
            "plan": self.current_plan,
            "next_billing": str(self.next_billing_date),
            "cancelled": self.cancelled
        }

# Production Worker Configuration

import asyncio
from temporalio.client import Client
from temporalio.worker import Worker
from temporalio.worker.workflow_sandbox import SandboxedWorkflowRunner
import logging

logger = logging.getLogger(__name__)


class WorkerConfig:
    """Configuration for Temporal workers."""
    
    def __init__(
        self,
        temporal_host: str = "localhost:7233",
        namespace: str = "default",
        task_queue: str = "orders",
        max_concurrent_activities: int = 100,
        max_concurrent_workflow_tasks: int = 100,
        sticky_queue_schedule_to_start_timeout: int = 10,
    ):
        self.temporal_host = temporal_host
        self.namespace = namespace
        self.task_queue = task_queue
        self.max_concurrent_activities = max_concurrent_activities
        self.max_concurrent_workflow_tasks = max_concurrent_workflow_tasks
        self.sticky_queue_schedule_to_start_timeout = sticky_queue_schedule_to_start_timeout


async def run_worker(config: WorkerConfig):
    """Run a Temporal worker."""
    
    # Connect to Temporal
    client = await Client.connect(
        config.temporal_host,
        namespace=config.namespace
    )
    
    # Create worker
    worker = Worker(
        client,
        task_queue=config.task_queue,
        
        # Register workflows
        workflows=[
            OrderWorkflow,
            OrderWithApprovalWorkflow,
            SubscriptionWorkflow,
            BatchOrderWorkflow,
        ],
        
        # Register activities
        activities=[
            reserve_inventory,
            release_inventory,
            charge_payment,
            refund_payment,
            create_shipment,
            send_notification,
            check_fraud,
            award_loyalty_points,
        ],
        
        # Concurrency limits
        max_concurrent_activities=config.max_concurrent_activities,
        max_concurrent_workflow_tasks=config.max_concurrent_workflow_tasks,
        
        # Sticky execution (workflow affinity)
        sticky_queue_schedule_to_start_timeout=timedelta(
            seconds=config.sticky_queue_schedule_to_start_timeout
        ),
    )
    
    logger.info(f"Starting worker on queue: {config.task_queue}")
    
    # Run worker until interrupted
    await worker.run()


async def run_activity_worker(config: WorkerConfig):
    """
    Run activity-only worker.
    
    Useful for scaling activities independently from workflows.
    """
    client = await Client.connect(
        config.temporal_host,
        namespace=config.namespace
    )
    
    worker = Worker(
        client,
        task_queue=config.task_queue,
        
        # Only activities, no workflows
        activities=[
            reserve_inventory,
            release_inventory,
            charge_payment,
            refund_payment,
            create_shipment,
            send_notification,
        ],
        
        max_concurrent_activities=config.max_concurrent_activities,
    )
    
    await worker.run()


# Entry point
if __name__ == "__main__":
    config = WorkerConfig(
        temporal_host=os.getenv("TEMPORAL_HOST", "localhost:7233"),
        namespace=os.getenv("TEMPORAL_NAMESPACE", "default"),
        task_queue=os.getenv("TASK_QUEUE", "orders"),
    )
    
    asyncio.run(run_worker(config))

# Production Client Usage

from temporalio.client import Client, WorkflowHandle
from temporalio.common import WorkflowIDReusePolicy
import asyncio


class OrderService:
    """
    Service for managing orders via Temporal workflows.
    """
    
    def __init__(self, temporal_client: Client):
        self.client = temporal_client
    
    async def create_order(self, input: OrderInput) -> str:
        """
        Start a new order workflow.
        
        Returns workflow ID for tracking.
        """
        # Use order ID as workflow ID for idempotency
        workflow_id = f"order-{input.order_id}"
        
        handle = await self.client.start_workflow(
            OrderWorkflow.run,
            input,
            id=workflow_id,
            task_queue="orders",
            
            # Prevent duplicate orders
            id_reuse_policy=WorkflowIDReusePolicy.REJECT_DUPLICATE,
            
            # Execution timeout (entire workflow)
            execution_timeout=timedelta(hours=24),
            
            # Run timeout (single run, before continue-as-new)
            run_timeout=timedelta(hours=1),
        )
        
        return workflow_id
    
    async def get_order_status(self, order_id: str) -> dict:
        """
        Query order status.
        
        Non-blocking - returns current state without waiting.
        """
        workflow_id = f"order-{order_id}"
        
        try:
            handle = self.client.get_workflow_handle(workflow_id)
            status = await handle.query(OrderWorkflow.get_status)
            tracking = await handle.query(OrderWorkflow.get_tracking_number)
            
            return {
                "order_id": order_id,
                "status": status,
                "tracking_number": tracking
            }
            
        except Exception as e:
            return {
                "order_id": order_id,
                "status": "unknown",
                "error": str(e)
            }
    
    async def wait_for_completion(
        self,
        order_id: str,
        timeout: timedelta = timedelta(minutes=30)
    ) -> OrderResult:
        """
        Wait for order to complete.
        
        Blocking - waits until workflow finishes.
        """
        workflow_id = f"order-{order_id}"
        
        handle = self.client.get_workflow_handle(workflow_id)
        
        try:
            result = await asyncio.wait_for(
                handle.result(),
                timeout=timeout.total_seconds()
            )
            return result
        except asyncio.TimeoutError:
            raise TimeoutError(f"Order {order_id} did not complete in {timeout}")
    
    async def cancel_order(self, order_id: str):
        """
        Cancel an order.
        
        Sends cancellation signal to workflow.
        """
        workflow_id = f"order-{order_id}"
        
        handle = self.client.get_workflow_handle(workflow_id)
        await handle.cancel()
    
    async def approve_order(self, order_id: str, approver: str):
        """Send approval signal to order waiting for approval."""
        workflow_id = f"order-{order_id}"
        
        handle = self.client.get_workflow_handle(workflow_id)
        await handle.signal(OrderWithApprovalWorkflow.approve, approver)


# FastAPI integration
from fastapi import FastAPI, HTTPException

app = FastAPI()
order_service: OrderService = None


@app.on_event("startup")
async def startup():
    global order_service
    client = await Client.connect("localhost:7233")
    order_service = OrderService(client)


@app.post("/orders")
async def create_order(request: OrderInput):
    """Create a new order."""
    workflow_id = await order_service.create_order(request)
    return {"workflow_id": workflow_id, "status": "started"}


@app.get("/orders/{order_id}")
async def get_order(order_id: str):
    """Get order status."""
    return await order_service.get_order_status(order_id)


@app.post("/orders/{order_id}/approve")
async def approve_order(order_id: str, approver: str):
    """Approve an order."""
    await order_service.approve_order(order_id, approver)
    return {"status": "approved"}

❌ WRONG: Non-deterministic workflow

@workflow.defn
class BadWorkflow:
    @workflow.run
    async def run(self, input: dict) -> dict:
        # DON'T: Use random in workflow
        if random.random() > 0.5:
            await self.do_something()
        
        # DON'T: Use current time directly
        timestamp = datetime.now()
        
        # DON'T: Use UUID directly
        id = str(uuid.uuid4())
        
        # DON'T: Make HTTP calls in workflow
        response = await httpx.get("https://api.example.com/data")
        
        # On replay, these will produce DIFFERENT values
        # causing non-determinism errors!


✅ CORRECT: Deterministic workflow

@workflow.defn
class GoodWorkflow:
    @workflow.run
    async def run(self, input: dict) -> dict:
        # DO: Use workflow.random()
        if workflow.random().random() > 0.5:
            await self.do_something()
        
        # DO: Use workflow.now()
        timestamp = workflow.now()
        
        # DO: Use workflow.uuid4()
        id = str(workflow.uuid4())
        
        # DO: Use activities for side effects
        response = await workflow.execute_activity(
            fetch_data,
            args=["https://api.example.com/data"],
            start_to_close_timeout=timedelta(minutes=1)
        )

❌ WRONG: Workflow that grows forever

@workflow.defn
class BadSubscriptionWorkflow:
    @workflow.run
    async def run(self, customer_id: str) -> None:
        while True:
            await workflow.sleep(timedelta(days=30))
            
            # Each iteration adds events to history, without bound
            # Replays get slower, and Temporal eventually terminates the run
            # (history limit: 51,200 events or 50 MB; warnings from 10K events)
            
            await workflow.execute_activity(
                charge_monthly,
                args=[customer_id],
                start_to_close_timeout=timedelta(minutes=5)
            )


✅ CORRECT: Use continue-as-new

@workflow.defn
class GoodSubscriptionWorkflow:
    @workflow.run
    async def run(self, customer_id: str, billing_count: int = 0) -> None:
        while billing_count < 12:  # One year max
            await workflow.sleep(timedelta(days=30))
            
            await workflow.execute_activity(
                charge_monthly,
                args=[customer_id],
                start_to_close_timeout=timedelta(minutes=5)
            )
            
            billing_count += 1
        
        # Continue with fresh history
        workflow.continue_as_new(args=[customer_id, 0])

❌ WRONG: No timeouts on activities

@workflow.defn
class BadWorkflow:
    @workflow.run
    async def run(self, input: dict) -> dict:
        # No timeout - the SDK rejects this (start_to_close or schedule_to_close
        # is required); an over-generous timeout is the real-world version of this bug
        result = await workflow.execute_activity(
            call_external_api,
            args=[input]
        )


✅ CORRECT: Always set appropriate timeouts

@workflow.defn
class GoodWorkflow:
    @workflow.run
    async def run(self, input: dict) -> dict:
        result = await workflow.execute_activity(
            call_external_api,
            args=[input],
            
            # Schedule-to-start: Time in task queue
            schedule_to_start_timeout=timedelta(minutes=5),
            
            # Start-to-close: Time for activity to complete
            start_to_close_timeout=timedelta(minutes=10),
            
            # Schedule-to-close: Total time from schedule to completion
            schedule_to_close_timeout=timedelta(minutes=15),
            
            # Heartbeat timeout for long activities
            heartbeat_timeout=timedelta(minutes=1),
            
            # Retry policy
            retry_policy=RetryPolicy(
                initial_interval=timedelta(seconds=1),
                maximum_interval=timedelta(minutes=1),
                maximum_attempts=5
            )
        )

try:
    credit_result = await workflow.execute_activity(
        check_credit,
        application_id,
        retry_policy=RetryPolicy(maximum_attempts=5)
    )
except ActivityError:
    # Infrastructure failure - need manual review
    self.status = 'manual_review_required'
    return

if credit_result['score'] < 600:
    # Business rejection
    return LoanResult(status='rejected', reason='credit_score')