orchestrata_ultimate.governance.escalation_handler.models

Reference › orchestrata_ultimate › governance › escalation_handler › models

The data types of the escalation handler, with no imports from the project. They are shared by handler, resolver, sla and the package initialiser.

Enums: EscalationStatus describes one hop (pending, in progress, completed, timeout, failed); ConflictType lists overlapping authority, boundary violation, resource conflict, policy conflict and ownership dispute; ResolutionStrategy is vote, priority or delegation; ResolutionStatus is pending, in progress, resolved, rejected or escalated. Dataclasses: Conflict (id, type, component, volks, description), EscalationHop (hop index, volk alias, status, timestamps, deadline, retry count, response), EscalationResult (the full history of an escalation with its hops and total time) and Resolution (id, workflow id, who resolved it, text and strategy). A _utc_now helper supplies timestamps. The kernel's PAM++ code has similar but separate types in kernel pam types.

At a glance

Class diagram

Dependency diagram

API

Classes

EscalationStatus (enum)

Bases: Enum · Line 20

Status of an escalation hop.

ConflictType (enum)

Bases: Enum · Line 30

Types of conflicts.

ResolutionStrategy (enum)

Bases: Enum · Line 40

Conflict resolution strategies.

ResolutionStatus (enum)

Bases: Enum · Line 48

Status of conflict resolution workflow.

Conflict (dataclass)

Decorators: dataclass · Line 59

Represents a conflict requiring resolution.

EscalationHop (dataclass)

Decorators: dataclass · Line 72

Represents a single hop in escalation path (REQ-001-US-005-002).

EscalationResult (dataclass)

Decorators: dataclass · Line 85

Result of multi-hop escalation (REQ-001-US-005-001).

Resolution (dataclass)

Decorators: dataclass · Line 99

Represents a conflict resolution.

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/governance/escalation_handler/models.py

# mypy: ignore-errors
"""Escalation data classes, enums, and the shared UTC time helper."""

from dataclasses import dataclass, field
from datetime import UTC, datetime
from enum import Enum
from typing import Any


def _utc_now() -> datetime:
    """Get current UTC time with timezone info."""
    return datetime.now(UTC)


# ================================================================
# Data Classes and Enums
# ================================================================


class EscalationStatus(Enum):
    """Status of an escalation hop."""

    pending = "pending"
    in_progress = "in_progress"
    completed = "completed"
    timeout = "timeout"
    failed = "failed"


class ConflictType(Enum):
    """Types of conflicts."""

    overlapping_authority = "overlapping_authority"
    boundary_violation = "boundary_violation"
    resource_conflict = "resource_conflict"
    policy_conflict = "policy_conflict"
    ownership_dispute = "ownership_dispute"


class ResolutionStrategy(Enum):
    """Conflict resolution strategies."""

    vote = "vote"
    priority = "priority"
    delegation = "delegation"


class ResolutionStatus(Enum):
    """Status of conflict resolution workflow."""

    pending = "pending"
    in_progress = "in_progress"
    resolved = "resolved"
    rejected = "rejected"
    escalated = "escalated"


@dataclass
class Conflict:
    """Represents a conflict requiring resolution."""

    conflict_id: str
    conflict_type: ConflictType
    component: str
    volks: list[str]
    description: str
    created_at: datetime = field(default_factory=_utc_now)
    metadata: dict[str, Any] = field(default_factory=dict)


@dataclass
class EscalationHop:
    """Represents a single hop in escalation path (REQ-001-US-005-002)."""

    hop_index: int
    volk_alias: str
    status: EscalationStatus
    timestamp: datetime
    response: str | None = None
    timeout_at: datetime | None = None
    retry_count: int = 0


@dataclass
class EscalationResult:
    """Result of multi-hop escalation (REQ-001-US-005-001)."""

    escalation_id: str
    conflict: Conflict
    hops: list[EscalationHop]
    resolved: bool
    resolution: str | None = None
    total_time_ms: int = 0
    created_at: datetime = field(default_factory=_utc_now)
    resolved_at: datetime | None = None


@dataclass
class Resolution:
    """Represents a conflict resolution."""

    resolution_id: str
    workflow_id: str
    resolved_by: str
    resolution_text: str
    strategy: ResolutionStrategy
    resolved_at: datetime = field(default_factory=_utc_now)
    metadata: dict[str, Any] = field(default_factory=dict)