The Complete Production SOP: How to Deploy and Run Temporal Workflow Orchestration & Durable Execution in Autonomous AI Agents

---
name: "temporal-workflow-deterministic-state"
description: "Write resilient, multi-day distributed workflows with automatic retries and durable state."
metadata:
  author: "Temporal Technologies"
  version: "1.0.0"
  runtime_compatibility:
    - "codex"
    - "claude-code"
    - "cursor"
    - "windsurf"
triggers:
  - "write temporal workflow"
  - "configure temporal activity retry"
  - "handle temporal workflow signal"
permissions:
  fs_read: true
  fs_write: "workspace"
  shell_exec: false
  net_access: false
---

# Temporal Workflow Orchestration & Durable Execution

## Execution SOP & Rules
1. Keep workflow functions strictly deterministic; never invoke network I/O, file systems, or random numbers in workflows.
2. Wrap side-effects and external API calls inside @activity.defn functions with retry policies.
3. Handle external events using @workflow.signal and expose read state using @workflow.query.
4. Use workflow.patched('change-id') when evolving running workflow definitions to prevent replay mismatches.

import os
from dataclasses import dataclass

@dataclass(frozen=True)
class TemporalSettings:
    target_host: str = os.getenv("TEMPORAL_TARGET_HOST", "localhost:7233")
    namespace: str = os.getenv("TEMPORAL_NAMESPACE", "default")
    task_queue: str = os.getenv("TEMPORAL_TASK_QUEUE", "order-fulfillment-v1")

settings = TemporalSettings()

from datetime import timedelta
from temporalio import workflow
from temporalio.common import RetryPolicy

with workflow.unsafe.imports_passed_through():
    from activities.payment import process_payment, PaymentInput

@workflow.defn
class OrderFulfillmentWorkflow:
    def __init__(self) -> None:
        self._payment_approved: bool = False
        self._cancelled: bool = False

    @workflow.signal
    def cancel_order(self) -> None:
        self._cancelled = True

    @workflow.query
    def is_approved(self) -> bool:
        return self._payment_approved

    @workflow.run
    async def run(self, order_id: str, amount_cents: int) -> str:
        if self._cancelled:
            return "CANCELLED_BEFORE_PAYMENT"

        retry_policy = RetryPolicy(
            initial_interval=timedelta(seconds=1),
            backoff_coefficient=2.0,
            maximum_interval=timedelta(seconds=30),
            maximum_attempts=5,
            non_retryable_error_types=["InvalidCreditCardError"]
        )

        payment_result = await workflow.execute_activity(
            process_payment,
            PaymentInput(order_id=order_id, amount_cents=amount_cents),
            start_to_close_timeout=timedelta(seconds=10),
            retry_policy=retry_policy,
        )

        self._payment_approved = payment_result.success
        return f"ORDER_COMPLETED: {order_id}"

import pytest
from temporalio.worker import WorkflowReplayer
from workflows.order_workflow import OrderFulfillmentWorkflow

@pytest.mark.asyncio
async def test_order_workflow_replay():
    replayer = WorkflowReplayer(workflows=[OrderFulfillmentWorkflow])
    # Validates deterministic replay without side-effects or network calls
    await replayer.replay_workflow(
        WorkflowReplayer.WorkflowHistory.from_json("order-101", open("tests/fixtures/order_101_history.json").read())
    )