Enterprise Approval Scenarios:

  Simple Approval          Multi-Level               Conditional Branch
  +----------+            +----------+              +----------+
  |Applicant |-> Approver |Applicant |-> Manager    |Applicant |-> Amount check
  +----------+            +----+-----+   -> Director+----+-----+
                               |         -> VP           |
                               v                         v
                          +----------+          +------------------+
                          |Countersign|         | <100K: Manager   |
                          |(parallel) |         | 100-500K: Dir    |
                          +----------+          | >500K: VP+CFO    |
                                                +------------------+

from __future__ import annotations
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from typing import Any


class ApprovalStatus(Enum):
    """Approval status"""
    DRAFT = "draft"
    PENDING = "pending"
    IN_REVIEW = "in_review"
    APPROVED = "approved"
    REJECTED = "rejected"
    WITHDRAWN = "withdrawn"
    ESCALATED = "escalated"
    EXPIRED = "expired"


class ApprovalAction(Enum):
    """Approval action"""
    SUBMIT = "submit"
    APPROVE = "approve"
    REJECT = "reject"
    WITHDRAW = "withdraw"
    ESCALATE = "escalate"
    RETURN = "return"
    COUNTERSIGN = "countersign"
    TIMEOUT = "timeout"


@dataclass
class ApprovalRequest:
    """Approval request"""
    request_id: str
    decision_id: str
    applicant: str
    domain: str
    title: str
    content: dict[str, Any]
    amount: float = 0.0
    priority: str = "normal"
    status: ApprovalStatus = ApprovalStatus.DRAFT
    current_level: int = 0
    approval_chain: list[ApprovalLevel] = field(default_factory=list)
    history: list[ApprovalEvent] = field(default_factory=list)
    created_at: datetime = field(default_factory=datetime.utcnow)
    updated_at: datetime = field(default_factory=datetime.utcnow)


@dataclass
class ApprovalLevel:
    """Approval level definition"""
    level: int
    approvers: list[str]
    mode: str = "any"            # any=or-sign, all=countersign
    timeout_hours: int = 48
    auto_escalate_to: str | None = None


@dataclass
class ApprovalEvent:
    """Approval event record"""
    event_id: str
    action: ApprovalAction
    actor: str
    comment: str = ""
    timestamp: datetime = field(default_factory=datetime.utcnow)
    metadata: dict[str, Any] = field(default_factory=dict)

class ApprovalStateMachine:
    """Approval state machine"""

    _transitions: dict[tuple[ApprovalStatus, ApprovalAction], ApprovalStatus] = {
        (ApprovalStatus.DRAFT, ApprovalAction.SUBMIT): ApprovalStatus.PENDING,
        (ApprovalStatus.PENDING, ApprovalAction.APPROVE): ApprovalStatus.IN_REVIEW,
        (ApprovalStatus.PENDING, ApprovalAction.REJECT): ApprovalStatus.REJECTED,
        (ApprovalStatus.PENDING, ApprovalAction.WITHDRAW): ApprovalStatus.WITHDRAWN,
        (ApprovalStatus.PENDING, ApprovalAction.ESCALATE): ApprovalStatus.ESCALATED,
        (ApprovalStatus.PENDING, ApprovalAction.TIMEOUT): ApprovalStatus.ESCALATED,
        (ApprovalStatus.PENDING, ApprovalAction.RETURN): ApprovalStatus.DRAFT,
        (ApprovalStatus.IN_REVIEW, ApprovalAction.APPROVE): ApprovalStatus.APPROVED,
        (ApprovalStatus.IN_REVIEW, ApprovalAction.REJECT): ApprovalStatus.REJECTED,
        (ApprovalStatus.IN_REVIEW, ApprovalAction.ESCALATE): ApprovalStatus.ESCALATED,
        (ApprovalStatus.ESCALATED, ApprovalAction.APPROVE): ApprovalStatus.APPROVED,
        (ApprovalStatus.ESCALATED, ApprovalAction.REJECT): ApprovalStatus.REJECTED,
        (ApprovalStatus.REJECTED, ApprovalAction.SUBMIT): ApprovalStatus.PENDING,
    }

    def can_transition(self, current: ApprovalStatus,
                       action: ApprovalAction) -> bool:
        return (current, action) in self._transitions

    def transition(self, request: ApprovalRequest,
                   action: ApprovalAction,
                   actor: str,
                   comment: str = "") -> ApprovalStatus:
        key = (request.status, action)
        if key not in self._transitions:
            raise ValueError(
                f"Invalid transition: {request.status.value} + {action.value}"
            )

        new_status = self._transitions[key]

        if (action == ApprovalAction.APPROVE and
                request.current_level < len(request.approval_chain) - 1):
            request.current_level += 1
            new_status = ApprovalStatus.PENDING

        request.status = new_status
        request.updated_at = datetime.utcnow()

        event = ApprovalEvent(
            event_id=f"evt-{len(request.history)}",
            action=action,
            actor=actor,
            comment=comment,
        )
        request.history.append(event)

        return new_status

    def get_available_actions(self, status: ApprovalStatus) -> list[ApprovalAction]:
        return [
            action for (s, action) in self._transitions
            if s == status
        ]

class ApprovalChainBuilder:
    """Approval chain builder"""

    def __init__(self):
        self._rules: list[dict] = []

    def add_rule(self, condition: str,
                 levels: list[ApprovalLevel]) -> ApprovalChainBuilder:
        self._rules.append({
            "condition": condition,
            "levels": levels,
        })
        return self

    def build(self, request: ApprovalRequest) -> list[ApprovalLevel]:
        """Build approval chain based on request content"""
        for rule in self._rules:
            if self._evaluate_condition(rule["condition"], request):
                return rule["levels"]

        return [ApprovalLevel(level=0, approvers=["default_approver"])]

    def _evaluate_condition(self, condition: str,
                            request: ApprovalRequest) -> bool:
        context = {
            "amount": request.amount,
            "domain": request.domain,
            "priority": request.priority,
        }
        try:
            return bool(eval(condition, {"__builtins__": {}}, context))
        except Exception:
            return False


# Usage example
chain_builder = (
    ApprovalChainBuilder()
    .add_rule(
        "amount > 500000",
        [
            ApprovalLevel(0, ["department_manager"], mode="any", timeout_hours=24),
            ApprovalLevel(1, ["vp_finance", "vp_ops"], mode="all", timeout_hours=48),
            ApprovalLevel(2, ["ceo"], mode="any", timeout_hours=72),
        ]
    )
    .add_rule(
        "amount > 100000",
        [
            ApprovalLevel(0, ["department_manager"], mode="any", timeout_hours=24),
            ApprovalLevel(1, ["vp_finance"], mode="any", timeout_hours=48),
        ]
    )
    .add_rule(
        "amount > 0",
        [
            ApprovalLevel(0, ["department_manager"], mode="any", timeout_hours=48),
        ]
    )
)

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


@workflow.defn
class ApprovalWorkflow:
    """Approval workflow"""

    def __init__(self):
        self._request: ApprovalRequest | None = None
        self._state_machine = ApprovalStateMachine()
        self._pending_action: ApprovalAction | None = None
        self._pending_actor: str = ""
        self._pending_comment: str = ""

    @workflow.run
    async def run(self, request_data: dict) -> dict:
        """Main workflow execution"""
        self._request = ApprovalRequest(**request_data)

        chain = await workflow.execute_activity(
            build_approval_chain,
            args=[request_data],
            start_to_close_timeout=timedelta(seconds=30),
        )
        self._request.approval_chain = [
            ApprovalLevel(**level) for level in chain
        ]

        self._state_machine.transition(
            self._request, ApprovalAction.SUBMIT,
            self._request.applicant
        )

        for level_idx, level in enumerate(self._request.approval_chain):
            self._request.current_level = level_idx

            await workflow.execute_activity(
                notify_approvers,
                args=[self._request.request_id,
                      level.approvers, level.level],
                start_to_close_timeout=timedelta(seconds=30),
            )

            result = await self._wait_for_approval(level)

            if result == "rejected":
                return self._build_result("rejected")
            elif result == "escalated":
                continue
            elif result == "withdrawn":
                return self._build_result("withdrawn")

        self._state_machine.transition(
            self._request, ApprovalAction.APPROVE,
            "system", "All levels approved"
        )

        await workflow.execute_activity(
            execute_approved_decision,
            args=[self._request.decision_id],
            start_to_close_timeout=timedelta(seconds=60),
            retry_policy=RetryPolicy(maximum_attempts=3),
        )

        return self._build_result("approved")

    async def _wait_for_approval(self, level: ApprovalLevel) -> str:
        if level.mode == "any":
            return await self._wait_any(level)
        else:
            return await self._wait_all(level)

    async def _wait_any(self, level: ApprovalLevel) -> str:
        """Or-sign: any single approval suffices"""
        timeout = timedelta(hours=level.timeout_hours)

        try:
            await workflow.wait_condition(
                lambda: self._pending_action is not None,
                timeout=timeout,
            )

            action = self._pending_action
            self._pending_action = None

            if action == ApprovalAction.APPROVE:
                self._state_machine.transition(
                    self._request, ApprovalAction.APPROVE,
                    self._pending_actor, self._pending_comment
                )
                return "approved"
            elif action == ApprovalAction.REJECT:
                self._state_machine.transition(
                    self._request, ApprovalAction.REJECT,
                    self._pending_actor, self._pending_comment
                )
                return "rejected"
            elif action == ApprovalAction.WITHDRAW:
                return "withdrawn"
            else:
                return "approved"

        except TimeoutError:
            if level.auto_escalate_to:
                self._state_machine.transition(
                    self._request, ApprovalAction.ESCALATE,
                    "system", f"Timeout after {level.timeout_hours}h"
                )
                return "escalated"
            else:
                return "rejected"

    async def _wait_all(self, level: ApprovalLevel) -> str:
        """Countersign: all approvers must approve"""
        remaining = set(level.approvers)
        timeout = timedelta(hours=level.timeout_hours)

        while remaining:
            try:
                await workflow.wait_condition(
                    lambda: self._pending_action is not None,
                    timeout=timeout,
                )

                action = self._pending_action
                actor = self._pending_actor
                self._pending_action = None

                if action == ApprovalAction.REJECT:
                    self._state_machine.transition(
                        self._request, ApprovalAction.REJECT,
                        actor, self._pending_comment
                    )
                    return "rejected"

                if action == ApprovalAction.APPROVE and actor in remaining:
                    remaining.discard(actor)

            except TimeoutError:
                return "escalated"

        return "approved"

    @workflow.signal
    async def approval_action(self, action: str, actor: str,
                               comment: str = ""):
        """Receive approval action via Signal"""
        self._pending_action = ApprovalAction(action)
        self._pending_actor = actor
        self._pending_comment = comment

    @workflow.query
    def get_status(self) -> dict:
        """Query approval status"""
        if self._request is None:
            return {"status": "not_started"}
        return {
            "request_id": self._request.request_id,
            "status": self._request.status.value,
            "current_level": self._request.current_level,
            "total_levels": len(self._request.approval_chain),
            "history": [
                {
                    "action": e.action.value,
                    "actor": e.actor,
                    "comment": e.comment,
                    "timestamp": e.timestamp.isoformat(),
                }
                for e in self._request.history
            ],
        }

    def _build_result(self, outcome: str) -> dict:
        return {
            "request_id": self._request.request_id,
            "outcome": outcome,
            "history_count": len(self._request.history),
        }

@activity.defn
async def build_approval_chain(request_data: dict) -> list[dict]:
    """Build approval chain"""
    request = ApprovalRequest(**request_data)
    chain = chain_builder.build(request)
    return [
        {
            "level": level.level,
            "approvers": level.approvers,
            "mode": level.mode,
            "timeout_hours": level.timeout_hours,
            "auto_escalate_to": level.auto_escalate_to,
        }
        for level in chain
    ]


@activity.defn
async def notify_approvers(request_id: str,
                            approvers: list[str],
                            level: int) -> None:
    """Notify approvers"""
    for approver in approvers:
        await notification_service.send(
            channel="email",
            recipient=approver,
            template="approval_pending",
            variables={
                "request_id": request_id,
                "level": level,
                "action_url": f"/approvals/{request_id}",
            },
        )


@activity.defn
async def execute_approved_decision(decision_id: str) -> dict:
    """Execute the approved decision"""
    result = await action_engine.execute(decision_id)
    return {"decision_id": decision_id, "execution_status": result.status}

syntax = "proto3";
package onto.approval.v1;

service ApprovalService {
    rpc SubmitApproval(SubmitRequest) returns (SubmitResponse);
    rpc ApproveOrReject(ActionRequest) returns (ActionResponse);
    rpc GetApprovalStatus(StatusRequest) returns (StatusResponse);
    rpc ListPendingApprovals(ListRequest) returns (ListResponse);
    rpc WithdrawApproval(WithdrawRequest) returns (WithdrawResponse);
}

message SubmitRequest {
    string decision_id = 1;
    string applicant = 2;
    string domain = 3;
    string title = 4;
    map<string, string> content = 5;
    double amount = 6;
    string priority = 7;
}

message ActionRequest {
    string request_id = 1;
    string action = 2;
    string actor = 3;
    string comment = 4;
}

message StatusResponse {
    string request_id = 1;
    string status = 2;
    int32 current_level = 3;
    int32 total_levels = 4;
    repeated HistoryEntry history = 5;
}

from temporalio.client import Client


class ApprovalServiceImpl:
    """Approval gRPC service"""

    def __init__(self, temporal_client: Client):
        self._client = temporal_client

    async def SubmitApproval(self, request, context):
        request_id = f"apr-{datetime.utcnow().strftime('%Y%m%d%H%M%S')}"

        handle = await self._client.start_workflow(
            ApprovalWorkflow.run,
            {
                "request_id": request_id,
                "decision_id": request.decision_id,
                "applicant": request.applicant,
                "domain": request.domain,
                "title": request.title,
                "content": dict(request.content),
                "amount": request.amount,
                "priority": request.priority,
            },
            id=f"approval-{request_id}",
            task_queue="approval-tasks",
        )

        return {"request_id": request_id, "workflow_id": handle.id}

    async def ApproveOrReject(self, request, context):
        handle = self._client.get_workflow_handle(
            f"approval-{request.request_id}"
        )
        await handle.signal(
            ApprovalWorkflow.approval_action,
            args=[request.action, request.actor, request.comment],
        )
        return {"status": "action_recorded"}

    async def GetApprovalStatus(self, request, context):
        handle = self._client.get_workflow_handle(
            f"approval-{request.request_id}"
        )
        status = await handle.query(ApprovalWorkflow.get_status)
        return status

class TimeoutManager:
    """Approval timeout manager"""

    @staticmethod
    async def setup_reminders(workflow_context,
                               level: ApprovalLevel,
                               request_id: str) -> None:
        total_hours = level.timeout_hours

        # Remind at 50%
        await asyncio.sleep(total_hours * 0.5 * 3600)
        for approver in level.approvers:
            await workflow.execute_activity(
                send_timeout_reminder,
                args=[request_id, approver, int(total_hours * 0.5)],
                start_to_close_timeout=timedelta(seconds=30),
            )

        # Remind again at 80%
        await asyncio.sleep(total_hours * 0.3 * 3600)
        for approver in level.approvers:
            await workflow.execute_activity(
                send_timeout_reminder,
                args=[request_id, approver, int(total_hours * 0.2)],
                start_to_close_timeout=timedelta(seconds=30),
            )

Approval Timeout Escalation Strategy:

  Timeline                                      Action
  +--- 0h --- Submit approval ------------------- Notify approvers
  |
  +--- 24h -- 50% reminder ---------------------- Send reminder
  |
  +--- 38h -- 80% reminder ---------------------- Urgent reminder
  |
  +--- 48h -- Timeout --------------------------- Auto-escalate
  |                                                |
  |              +---------------------------------+
  |              v
  |         Escalate to senior approver
  |         Reset timeout timer
  |
  +--- 96h -- Second timeout -------------------- Escalate to VP

Approval Kanban View:

  Pending (12)          In Review (5)        Completed (89)
  +----------------+   +----------------+   +----------------+
  | APR-0421       |   | APR-0415       |   | APR-0410  ok   |
  | Purchase $12K  |   | Budget $70K    |   | Travel $1.7K   |
  | Waiting: J.Mgr |   | L2/3 VP review |   | Approved       |
  | Remaining: 36h |   | Remaining: 12h |   |                |
  +----------------+   +----------------+   +----------------+
  | APR-0420       |   | APR-0413       |   | APR-0408  x    |
  | Contract $28K  |   | HR $110K       |   | Purchase $7K   |
  | Waiting: Dir   |   | L2/3 CFO csgn  |   | Rejected       |
  | Remaining: 22h |   | Remaining: 45h |   |                |
  +----------------+   +----------------+   +----------------+

class DecisionApprovalIntegration:
    """Decision engine and approval engine integration"""

    def __init__(self, decision_engine, approval_service):
        self._decision = decision_engine
        self._approval = approval_service

    async def evaluate_with_approval(self, context) -> dict:
        """Evaluate decision and trigger approval if needed"""
        result = self._decision.evaluate(context)

        needs_approval = self._check_approval_needed(result, context)

        if not needs_approval:
            return {
                "decision": result.decision,
                "status": "auto_approved",
                "approval_required": False,
            }

        approval_result = await self._approval.SubmitApproval({
            "decision_id": context.context_id,
            "applicant": context.metadata.get("applicant", "system"),
            "domain": context.domain,
            "title": f"Decision approval: {context.context_id}",
            "content": context.inputs,
            "amount": context.inputs.get("amount", 0),
        })

        return {
            "decision": result.decision,
            "status": "pending_approval",
            "approval_required": True,
            "approval_id": approval_result["request_id"],
        }

    def _check_approval_needed(self, result, context) -> bool:
        if result.confidence < 0.7:
            return True
        if context.inputs.get("amount", 0) > 10000:
            return True
        if context.domain in ("credit", "compliance", "hr"):
            return True
        return False

Temporal Reliability Mechanisms:

  +-------------------------------------+
  |           Temporal Server            |
  |                                     |
  |  +-----------+  +---------------+   |
  |  | Workflow   |  | Event History |   |
  |  | Execution  |  | (durable)    |   |
  |  +-----------+  +---------------+   |
  |                                     |
  |  Features:                          |
  |  - Auto-retry failed Activities     |
  |  - Auto-recover after worker crash  |
  |  - Complete event history audit     |
  |  - Versioned Workflow upgrades      |
  +-------------------------------------+

# Scenario: Credit decision triggers multi-level approval

# 1. Decision engine evaluation
context = (
    DecisionContextBuilder("credit")
    .with_inputs(
        credit_score=620,
        amount=300000,
        debt_ratio=0.45,
        applicant="user-12345",
    )
    .build()
)

# 2. Evaluate and trigger approval
integration = DecisionApprovalIntegration(decision_engine, approval_service)
result = await integration.evaluate_with_approval(context)
# {
#   "decision": "conditional_approve",
#   "status": "pending_approval",
#   "approval_required": True,
#   "approval_id": "apr-20260324103000",
# }

# 3. Approver action
await approval_service.ApproveOrReject({
    "request_id": "apr-20260324103000",
    "action": "approve",
    "actor": "department_manager",
    "comment": "Good credit history, approved",
})

# 4. Query status
status = await approval_service.GetApprovalStatus({
    "request_id": "apr-20260324103000",
})
# {
#   "status": "pending",
#   "current_level": 1,
#   "total_levels": 2,
#   "history": [...]
# }