┌─────────────────────────────────────────────────┐
│                  Agent Runtime Layer: Agent Runtime          │
│  ┌──────────┐  ┌──────────┐  ┌──────────┐      │
│  │ Action   │  │ Pipeline │  │ Agent    │      │
│  │ Workflow │  │ Workflow │  │ Workflow │      │
│  └────┬─────┘  └────┬─────┘  └────┬─────┘      │
│       │              │              │            │
│  ┌────┴──────────────┴──────────────┴────┐      │
│  │         Temporal Python SDK            │      │
│  │    (Worker / Client / Activities)      │      │
│  └────────────────┬──────────────────────┘      │
└───────────────────┼──────────────────────────────┘
                    │ gRPC
┌───────────────────┼──────────────────────────────┐
│  Temporal Server  │                              │
│  ┌─────────┐  ┌──┴──────┐  ┌──────────┐        │
│  │Frontend │  │ History │  │ Matching │        │
│  │ Service │  │ Service │  │ Service  │        │
│  └─────────┘  └─────────┘  └──────────┘        │
│                     │                            │
│  ┌──────────────────┴───────────────────┐       │
│  │    Persistence (PostgreSQL)           │       │
│  └──────────────────────────────────────┘       │
└──────────────────────────────────────────────────┘

Client SDK ──gRPC──→ Frontend Service
                        │
                        ├── Rate Limiting (token bucket)
                        ├── Request Validation
                        ├── Namespace Routing
                        └── Forward to History / Matching Service

# coomia-dip Namespace management
from temporalio.client import Client

async def create_world_namespace(world_id: str) -> None:
    client = await Client.connect("temporal-server:7233")
    # Each World uses an independent Namespace for isolation
    # Namespace naming: coomia-dip-{world_id}
    await client.operator_service.create_namespace(
        name=f"coomia-dip-{world_id}",
        retention_period=timedelta(days=30),  # 30-day workflow history retention
    )

Workflow Execution
    │
    ├── Shard 1 ──→ History Service Instance A
    ├── Shard 2 ──→ History Service Instance B
    ├── Shard 3 ──→ History Service Instance A  (multiple shards per instance)
    └── Shard N ──→ History Service Instance C

-- History Event storage structure
CREATE TABLE executions_v2 (
    shard_id        INT NOT NULL,
    namespace_id    BINARY(16) NOT NULL,
    workflow_id     VARCHAR(255) NOT NULL,
    run_id          BINARY(16) NOT NULL,
    event_id        BIGINT NOT NULL,
    event_type      INT NOT NULL,
    event_payload   BLOB NOT NULL,
    PRIMARY KEY (shard_id, namespace_id, workflow_id, run_id, event_id)
);

Worker Poll ──→ Matching Service ──→ Return Workflow/Activity Task

Workflow Task Queue:
  ┌─────────────────────────────────────┐
  │ Task 1: WF-abc, Decision needed     │ → Worker A polls
  │ Task 2: WF-def, Decision needed     │ → Worker B polls
  │ Task 3: WF-ghi, Decision needed     │ → Worker A polls (round-robin)
  └─────────────────────────────────────┘

Activity Task Queue:
  ┌─────────────────────────────────────┐
  │ Task 1: Execute ApprovalActivity    │ → Worker C polls
  │ Task 2: Execute NotifyActivity      │ → Worker D polls
  └─────────────────────────────────────┘

import asyncio
from temporalio.client import Client
from temporalio.worker import Worker

from coomia-dip.workflows.action_workflow import ActionApprovalWorkflow
from coomia-dip.activities.action_activities import (
    validate_action,
    execute_action,
    notify_approvers,
    record_audit_log,
)

async def run_worker():
    client = await Client.connect("temporal-server:7233")

    worker = Worker(
        client,
        task_queue="coomia-dip-action-queue",
        workflows=[ActionApprovalWorkflow],
        activities=[
            validate_action,
            execute_action,
            notify_approvers,
            record_audit_log,
        ],
        max_concurrent_workflow_tasks=100,
        max_concurrent_activities=50,
        max_cached_workflows=500,
    )

    await worker.run()

if __name__ == "__main__":
    asyncio.run(run_worker())

Event History for Workflow "action-approval-123":

  Event 1:  WorkflowExecutionStarted
  Event 2:  WorkflowTaskScheduled
  Event 3:  WorkflowTaskStarted
  Event 4:  WorkflowTaskCompleted
  Event 5:  ActivityTaskScheduled (validate_action)
  Event 6:  ActivityTaskStarted
  Event 7:  ActivityTaskCompleted (result: valid)
  Event 8:  WorkflowTaskScheduled
  Event 9:  WorkflowTaskStarted
  Event 10: WorkflowTaskCompleted
  Event 11: TimerStarted (wait for approval, 24h)
  Event 12: SignalExternalWorkflowExecutionInitiated
  ...
  Event 25: WorkflowExecutionCompleted

# Workflow code (all decision logic re-executes during Replay)
@workflow.defn
class ActionApprovalWorkflow:
    @workflow.run
    async def run(self, request: ActionRequest) -> ActionResult:
        # Step 1: Validate — During Replay, skips Activity execution, uses recorded result
        validation = await workflow.execute_activity(
            validate_action,
            request,
            start_to_close_timeout=timedelta(seconds=30),
        )

        # Step 2: Notify approvers
        await workflow.execute_activity(
            notify_approvers,
            request.approvers,
            start_to_close_timeout=timedelta(seconds=10),
        )

        # Step 3: Wait for approval (may wait days)
        # During Replay, if Signal Event exists, returns result immediately
        approval = await workflow.wait_condition(
            lambda: self._approval_decision is not None,
            timeout=timedelta(hours=24),
        )

        if self._approval_decision == "approved":
            result = await workflow.execute_activity(
                execute_action,
                request,
                start_to_close_timeout=timedelta(minutes=5),
            )
            return ActionResult(status="completed", data=result)
        else:
            return ActionResult(status="rejected")

# Large Event History optimization: use Continue-As-New
@workflow.defn
class LongRunningPipelineWorkflow:
    @workflow.run
    async def run(self, state: PipelineState) -> PipelineResult:
        while not state.is_complete:
            batch = await workflow.execute_activity(
                process_next_batch,
                state,
                start_to_close_timeout=timedelta(minutes=10),
            )
            state.update(batch)
            state.iterations += 1

            # Every 1000 iterations, use Continue-As-New to reset Event History
            if state.iterations % 1000 == 0:
                workflow.continue_as_new(state)

        return PipelineResult(state)

# ❌ Code violating Determinism

@workflow.defn
class BadWorkflow:
    @workflow.run
    async def run(self, request: dict) -> str:
        # ❌ Using system time (different during Replay)
        if datetime.now() > some_deadline:
            pass

        # ❌ Using random numbers (different results during Replay)
        if random.random() > 0.5:
            pass

        # ❌ Using UUID (generates different values during Replay)
        task_id = str(uuid.uuid4())

        # ❌ Direct I/O operations (file/network/database)
        data = requests.get("http://api.example.com/data")

        # ❌ Using mutable global state
        global_counter += 1

# ✅ Correct Deterministic code

@workflow.defn
class GoodWorkflow:
    @workflow.run
    async def run(self, request: dict) -> str:
        # ✅ Use Temporal-provided time
        now = workflow.now()

        # ✅ Use Temporal-provided random (Replay-safe)
        value = workflow.random().random()

        # ✅ Use Activity to generate UUID
        task_id = await workflow.execute_activity(
            generate_task_id,
            start_to_close_timeout=timedelta(seconds=5),
        )

        # ✅ I/O operations belong in Activities
        data = await workflow.execute_activity(
            fetch_data,
            "http://api.example.com/data",
            start_to_close_timeout=timedelta(seconds=30),
        )

@workflow.defn
class EvolvingWorkflow:
    @workflow.run
    async def run(self, request: ActionRequest) -> ActionResult:
        # Versioning: old Workflows take old path, new Workflows take new path
        if workflow.patched("add-risk-check"):
            # New version: add risk check step
            risk = await workflow.execute_activity(
                check_risk_level,
                request,
                start_to_close_timeout=timedelta(seconds=30),
            )
            if risk.level == "HIGH":
                request.require_extra_approval = True

        # Subsequent logic unchanged
        validation = await workflow.execute_activity(
            validate_action,
            request,
            start_to_close_timeout=timedelta(seconds=30),
        )

from temporalio.common import RetryPolicy

# coomia-dip standard retry policy
STANDARD_RETRY = RetryPolicy(
    initial_interval=timedelta(seconds=1),      # First retry interval
    backoff_coefficient=2.0,                      # Exponential backoff coefficient
    maximum_interval=timedelta(minutes=5),        # Maximum retry interval
    maximum_attempts=10,                           # Maximum retry attempts
    non_retryable_error_types=[                   # Non-retryable error types
        "ValidationError",
        "PermissionDeniedError",
        "NotFoundError",
    ],
)

# Using the retry policy
result = await workflow.execute_activity(
    execute_action,
    request,
    start_to_close_timeout=timedelta(minutes=5),
    retry_policy=STANDARD_RETRY,
)

wait(N) = min(initial_interval × backoff_coefficient^(N-1), maximum_interval)

Example (initial=1s, coefficient=2.0, max=300s):
  Retry 1: min(1 × 2^0, 300) = 1s
  Retry 2: min(1 × 2^1, 300) = 2s
  Retry 3: min(1 × 2^2, 300) = 4s
  Retry 4: min(1 × 2^3, 300) = 8s
  Retry 5: min(1 × 2^4, 300) = 16s
  Retry 6: min(1 × 2^5, 300) = 32s
  Retry 7: min(1 × 2^6, 300) = 64s
  Retry 8: min(1 × 2^7, 300) = 128s
  Retry 9: min(1 × 2^8, 300) = 256s
  Retry 10: min(1 × 2^9, 300) = 300s (capped)

Total wait time ≈ 811s ≈ 13.5 minutes

# Scenario 1: External API calls (may be temporarily unavailable)
EXTERNAL_API_RETRY = RetryPolicy(
    initial_interval=timedelta(seconds=2),
    backoff_coefficient=3.0,
    maximum_interval=timedelta(minutes=10),
    maximum_attempts=15,
    non_retryable_error_types=["AuthenticationError"],
)

# Scenario 2: Database operations (transient connection issues)
DB_RETRY = RetryPolicy(
    initial_interval=timedelta(milliseconds=100),
    backoff_coefficient=2.0,
    maximum_interval=timedelta(seconds=30),
    maximum_attempts=5,
)

# Scenario 3: Idempotent writes (safe to retry)
IDEMPOTENT_WRITE_RETRY = RetryPolicy(
    initial_interval=timedelta(seconds=1),
    backoff_coefficient=2.0,
    maximum_interval=timedelta(minutes=1),
    maximum_attempts=20,
)

# Scenario 4: Non-idempotent operations (caution required)
NON_IDEMPOTENT_RETRY = RetryPolicy(
    initial_interval=timedelta(seconds=5),
    backoff_coefficient=1.5,
    maximum_interval=timedelta(seconds=30),
    maximum_attempts=3,
)

@activity.defn
async def process_large_dataset(dataset_id: str) -> ProcessResult:
    """Process large dataset with progress reporting via Heartbeat"""
    records = await load_dataset(dataset_id)
    processed = 0

    for batch in chunk(records, size=1000):
        result = await process_batch(batch)
        processed += len(batch)

        # Send Heartbeat with progress info
        activity.heartbeat({"processed": processed, "total": len(records)})

    return ProcessResult(total_processed=processed)

# Configure Heartbeat timeout in Workflow
result = await workflow.execute_activity(
    process_large_dataset,
    dataset_id,
    start_to_close_timeout=timedelta(hours=2),
    heartbeat_timeout=timedelta(seconds=30),  # No heartbeat for 30s = Activity failed
    retry_policy=STANDARD_RETRY,
)

Create Order → Deduct Inventory → Send Notification → Update Audit Log
    ✅              ✅                  ❌ (failed!)
                                       ↓
              Need rollback: Restore Inventory → Cancel Order

from dataclasses import dataclass, field

@dataclass
class SagaStep:
    """A single step in a Saga"""
    name: str
    action: str  # Activity function name
    compensation: str  # Compensation Activity function name
    args: dict = field(default_factory=dict)


@workflow.defn
class SagaWorkflow:
    """Generic Saga Workflow"""

    @workflow.run
    async def run(self, steps: list[SagaStep]) -> dict:
        compensations: list[tuple[str, dict]] = []

        for step in steps:
            try:
                # Execute forward operation
                result = await workflow.execute_activity(
                    step.action,
                    step.args,
                    start_to_close_timeout=timedelta(minutes=5),
                    retry_policy=STANDARD_RETRY,
                )
                # Record compensation (LIFO order)
                compensations.append((step.compensation, {
                    **step.args,
                    "forward_result": result,
                }))

            except Exception as e:
                workflow.logger.error(
                    f"Step '{step.name}' failed: {e}. "
                    f"Starting compensation for {len(compensations)} completed steps."
                )
                # Execute all compensations in reverse
                await self._compensate(compensations)
                raise

        return {"status": "completed", "steps": len(steps)}

    async def _compensate(self, compensations: list[tuple[str, dict]]) -> None:
        """Execute compensations in reverse order"""
        errors = []
        for comp_activity, comp_args in reversed(compensations):
            try:
                await workflow.execute_activity(
                    comp_activity,
                    comp_args,
                    start_to_close_timeout=timedelta(minutes=5),
                    retry_policy=RetryPolicy(
                        initial_interval=timedelta(seconds=2),
                        maximum_attempts=5,
                    ),
                )
            except Exception as e:
                errors.append(f"Compensation '{comp_activity}' failed: {e}")
                workflow.logger.error(f"Compensation failed: {e}")

        if errors:
            raise CompensationError(errors)

@workflow.defn
class ActionExecutionWorkflow:
    """coomia-dip Action execution workflow with Saga compensation"""

    @workflow.run
    async def run(self, action_request: ActionRequest) -> ActionResult:
        compensations = []

        try:
            # Step 1: Validate Action parameters
            await workflow.execute_activity(
                validate_action_params,
                action_request,
                start_to_close_timeout=timedelta(seconds=30),
            )

            # Step 2: Acquire optimistic lock
            lock = await workflow.execute_activity(
                acquire_lock,
                action_request.target_object_id,
                start_to_close_timeout=timedelta(seconds=10),
            )
            compensations.append(("release_lock", lock))

            # Step 3: Apply data mutation
            mutation_result = await workflow.execute_activity(
                apply_mutation,
                action_request,
                start_to_close_timeout=timedelta(minutes=2),
            )
            compensations.append(("rollback_mutation", mutation_result))

            # Step 4: Trigger webhooks
            await workflow.execute_activity(
                trigger_webhooks,
                action_request,
                start_to_close_timeout=timedelta(seconds=30),
            )

            # Step 5: Record audit log
            await workflow.execute_activity(
                record_audit,
                action_request,
                start_to_close_timeout=timedelta(seconds=10),
            )

            # Step 6: Release lock
            await workflow.execute_activity(
                release_lock,
                lock,
                start_to_close_timeout=timedelta(seconds=10),
            )
            compensations = [c for c in compensations if c[0] != "release_lock"]

            return ActionResult(status="success", data=mutation_result)

        except Exception as e:
            # Saga compensation
            for comp_name, comp_args in reversed(compensations):
                try:
                    await workflow.execute_activity(
                        comp_name,
                        comp_args,
                        start_to_close_timeout=timedelta(minutes=1),
                        retry_policy=RetryPolicy(maximum_attempts=3),
                    )
                except Exception as comp_error:
                    workflow.logger.error(
                        f"Compensation {comp_name} failed: {comp_error}"
                    )

            return ActionResult(status="failed", error=str(e))

@workflow.defn
class ApprovalWorkflow:
    def __init__(self):
        self._approval_decision: str | None = None
        self._approval_comment: str = ""

    @workflow.signal
    async def approve(self, comment: str = "") -> None:
        """Approval signal"""
        self._approval_decision = "approved"
        self._approval_comment = comment

    @workflow.signal
    async def reject(self, comment: str = "") -> None:
        """Rejection signal"""
        self._approval_decision = "rejected"
        self._approval_comment = comment

    @workflow.run
    async def run(self, request: ApprovalRequest) -> ApprovalResult:
        # Notify approvers
        await workflow.execute_activity(
            send_approval_notification,
            request,
            start_to_close_timeout=timedelta(seconds=30),
        )

        # Wait for approval signal (max 24 hours)
        try:
            await workflow.wait_condition(
                lambda: self._approval_decision is not None,
                timeout=timedelta(hours=24),
            )
        except asyncio.TimeoutError:
            return ApprovalResult(status="timeout")

        return ApprovalResult(
            status=self._approval_decision,
            comment=self._approval_comment,
        )

@workflow.defn
class PipelineWorkflow:
    def __init__(self):
        self._progress = 0
        self._current_stage = "initializing"
        self._error_count = 0

    @workflow.query
    def get_progress(self) -> dict:
        """Query pipeline execution progress"""
        return {
            "progress": self._progress,
            "stage": self._current_stage,
            "errors": self._error_count,
        }

    @workflow.run
    async def run(self, pipeline_config: PipelineConfig) -> PipelineResult:
        stages = ["extract", "transform", "validate", "load"]

        for i, stage in enumerate(stages):
            self._current_stage = stage
            self._progress = int((i / len(stages)) * 100)

            try:
                await workflow.execute_activity(
                    f"execute_{stage}",
                    pipeline_config,
                    start_to_close_timeout=timedelta(minutes=30),
                )
            except Exception as e:
                self._error_count += 1
                raise

        self._progress = 100
        self._current_stage = "completed"
        return PipelineResult(status="success")

# Sending Signal from FastAPI endpoint
@app.post("/api/v1/actions/{action_id}/approve")
async def approve_action(action_id: str, body: ApprovalBody):
    client = await Client.connect("temporal-server:7233")
    handle = client.get_workflow_handle(f"action-approval-{action_id}")

    # Send Signal
    await handle.signal(ApprovalWorkflow.approve, body.comment)
    return {"status": "signal_sent"}


# Query Workflow state
@app.get("/api/v1/pipelines/{pipeline_id}/progress")
async def get_pipeline_progress(pipeline_id: str):
    client = await Client.connect("temporal-server:7233")
    handle = client.get_workflow_handle(f"pipeline-{pipeline_id}")

    # Query (synchronous, returns immediately)
    progress = await handle.query(PipelineWorkflow.get_progress)
    return progress

@workflow.defn
class BatchProcessingWorkflow:
    @workflow.run
    async def run(self, batch_config: BatchConfig) -> BatchResult:
        # Split large batch into sub-batches
        sub_batches = split_into_sub_batches(batch_config, chunk_size=1000)

        # Start Child Workflows in parallel
        handles = []
        for i, sub_batch in enumerate(sub_batches):
            handle = await workflow.start_child_workflow(
                SubBatchWorkflow.run,
                sub_batch,
                id=f"sub-batch-{batch_config.id}-{i}",
                parent_close_policy=ParentClosePolicy.REQUEST_CANCEL,
            )
            handles.append(handle)

        # Wait for all Child Workflows to complete
        results = await asyncio.gather(
            *[h.result() for h in handles],
            return_exceptions=True,
        )

        successful = sum(1 for r in results if not isinstance(r, Exception))
        failed = sum(1 for r in results if isinstance(r, Exception))

        return BatchResult(
            total=len(sub_batches),
            successful=successful,
            failed=failed,
        )

@workflow.defn
class ETLPipelineWorkflow:
    @workflow.run
    async def run(self, config: ETLConfig) -> ETLResult:
        # Stage 1: Extract
        extracted = await workflow.execute_child_workflow(
            ExtractWorkflow.run,
            config.source,
            id=f"etl-extract-{config.id}",
        )

        # Stage 2: Transform (depends on Extract result)
        transformed = await workflow.execute_child_workflow(
            TransformWorkflow.run,
            TransformInput(data=extracted, rules=config.transform_rules),
            id=f"etl-transform-{config.id}",
        )

        # Stage 3: Load (depends on Transform result)
        loaded = await workflow.execute_child_workflow(
            LoadWorkflow.run,
            LoadInput(data=transformed, target=config.target),
            id=f"etl-load-{config.id}",
        )

        return ETLResult(
            records_extracted=extracted.count,
            records_transformed=transformed.count,
            records_loaded=loaded.count,
        )

worker = Worker(
    client,
    task_queue="coomia-dip-action-queue",
    workflows=[ActionApprovalWorkflow],
    activities=[validate_action, execute_action],
    # Concurrency control
    max_concurrent_workflow_tasks=100,    # Concurrent Workflow Tasks
    max_concurrent_activities=50,         # Concurrent Activities
    max_cached_workflows=500,             # Cached Workflow instances (reduces Replay)
    # Sticky Queue (reduces Replay frequency)
    max_concurrent_workflow_task_polls=5,
    max_concurrent_activity_task_polls=5,
)

# Key metrics
TEMPORAL_METRICS = {
    "workflow_task_schedule_to_start_latency":  "Workflow Task scheduling latency",
    "activity_schedule_to_start_latency":       "Activity scheduling latency",
    "workflow_endtoend_latency":                "Workflow end-to-end latency",
    "workflow_failed":                          "Workflow failure count",
    "activity_execution_failed":                "Activity execution failure count",
}