orchestrata_ultimate.kernel.pam.escalation

Reference › orchestrata_ultimate › kernel › pam › escalation

Escalation management for PAM++, SPEC-001 stories US-003 and US-005. When the enforcer's checks produce a conflict between volks, this module decides who must settle it and tracks the case until it is resolved or times out. The pam package imports it, the value types are in types, and the multi-hop part uses AdvancedEscalationHandler from the governance escalation handler.

EscalationManager holds the open escalations by id. detect_conflict takes a ConflictContext and asks Prolog escalation_required/1 for its conflict type; it returns a Conflict with a generated id, or None when no escalation is needed or the query fails. get_escalation_path asks pam_escalation_path/2 and escalation_priority/2 and builds an EscalationPath of numbered steps; with no answer it falls back to a single step to the orchestrator alias U:Ox at priority 1. escalate assigns an id, sets a timeout (default 3600 seconds), stores a pending EscalationResult, marks the first step as notified and logs it to the graph sink when present. resolve marks an escalation resolved with the resolution text, and timeout_stale marks pending or in-progress escalations past their deadline as timed out and returns their ids. get_escalation and get_all_escalations read the store.

MultiHopEscalationManager adds multi-hop routing on top: route_escalation returns the authority of the first step of an existing escalation (single-hop, for backward compatibility), and multi_hop_escalate hands a conflict and a path of volk aliases to the governance AdvancedEscalationHandler, starting at the first alias or orchestrator when the path is empty.

At a glance

Class diagram

Dependency diagram

API

Classes

EscalationManager

Decorators: traces(spec='001', story='US-003', reqs=['REQ-001-US-003-021', 'REQ-001-US-003-023', 'REQ-001-US-003-024']) · Line 48

Escalation Manager for conflict resolution.

_EscalationLookup (protocol)

Bases: Protocol · Decorators: runtime_checkable · Line 334

Structural shape route_escalation actually needs.

MultiHopEscalationManager

Decorators: traces(spec='001', story='US-003', reqs=['REQ-001-US-003-023', 'REQ-001-US-003-028', 'REQ-001-US-005-014', 'REQ-001-US-… · Line 364

Enhanced escalation manager with multi-hop support (US-005).

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/kernel/pam/escalation.py

# mypy: ignore-errors
"""Escalation management for PAM++ (EscalationManager, MultiHopEscalationManager).

Decomposed from kernel/pam_enforcer.py. Handles conflict detection, escalation
routing, resolution tracking, and advanced multi-hop escalation.
"""

import asyncio
import logging
import uuid
from datetime import timedelta
from typing import Any, Protocol, runtime_checkable

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.core.protocols import IGovernancePrologBridge

# Import for Advanced Escalation Integration (US-005)
# Kept at module top to avoid circular dependency
from orchestrata_ultimate.governance.escalation_handler import (
    AdvancedEscalationHandler,
)
from orchestrata_ultimate.governance.escalation_handler import (
    Conflict as AdvancedConflict,
)
from orchestrata_ultimate.kernel.pam.types import (
    Conflict,
    ConflictContext,
    EscalationPath,
    EscalationResult,
    EscalationStatus,
    EscalationStep,
    Resolution,
    _utc_now,
)

logger = logging.getLogger(__name__)


@traces(
    spec="001",
    story="US-003",
    reqs=[
        "REQ-001-US-003-021",
        "REQ-001-US-003-023",
        "REQ-001-US-003-024",
    ],
)
class EscalationManager:
    """Escalation Manager for conflict resolution.

    Handles conflict detection, escalation routing, and resolution tracking.
    """

    def __init__(
        self,
        prolog_bridge: IGovernancePrologBridge,
        graph_audit: Any | None = None,
        default_timeout_seconds: int = 3600,
        escalations: dict[str, EscalationResult] | None = None,
    ) -> None:
        """Initialize escalation manager.

        Args:
            prolog_bridge: IGovernancePrologBridge for rule queries
            graph_audit: Graph audit trail (optional)
            default_timeout_seconds: Default escalation timeout
            escalations: Initial escalations dict (optional)
        """
        self._prolog = prolog_bridge
        self._graph = graph_audit
        self._default_timeout = default_timeout_seconds
        self._escalations: dict[str, EscalationResult] = (
            escalations if escalations is not None else {}
        )

    @traces(
        spec="001",
        story="US-003",
        reqs=[
            "REQ-001-US-003-021",
            "REQ-001-US-005-001",
            "REQ-001-US-005-002",
            "REQ-001-US-005-003",
        ],
    )
    async def detect_conflict(self, context: ConflictContext) -> Conflict | None:
        """Detect if context represents a conflict (REQ-001-US-003-021).

        Args:
            context: Conflict context

        Returns:
            Conflict if detected, None otherwise
        """
        # Query Prolog to check if escalation needed (async)
        conflict_type = context.conflict_type.value
        query = f"pam:escalation_required({conflict_type})"

        loop = asyncio.get_running_loop()
        try:
            needs_esc = await loop.run_in_executor(None, self._prolog.query_bool, query)
        except Exception as e:
            logger.error(f"Prolog query failed for conflict detection: {e}")
            needs_esc = False

        if not needs_esc:
            return None

        conflict_id = f"conflict-{uuid.uuid4().hex[:8]}"

        return Conflict(
            conflict_id=conflict_id,
            conflict_type=context.conflict_type,
            context=context,
        )

    @traces(
        spec="001",
        story="US-003",
        reqs=["REQ-001-US-003-022", "REQ-001-US-005-011"],
    )
    async def get_escalation_path(self, conflict: Conflict) -> EscalationPath:
        """Get escalation path for conflict (REQ-001-US-003-022).

        Args:
            conflict: Conflict to escalate

        Returns:
            EscalationPath with steps and priority
        """
        conflict_type = conflict.conflict_type.value
        loop = asyncio.get_running_loop()

        # Query Prolog for escalation path (async)
        path_query = f"pam:pam_escalation_path({conflict_type}, Path)"
        try:
            path_result = await loop.run_in_executor(
                None, self._prolog.query_one, path_query
            )
        except Exception as e:
            logger.error(f"Prolog query failed for escalation path: {e}")
            path_result = None

        if not path_result:
            # Default to Orchestrator
            return EscalationPath(
                steps=[EscalationStep(level=1, authority="U:Ox")],
                final_authority="U:Ox",
                priority=1,
            )

        # Parse path from Prolog list
        path_list = path_result.get("Path", ["U:Ox"])

        steps = [
            EscalationStep(level=idx + 1, authority=auth)
            for idx, auth in enumerate(path_list)
        ]

        # Get priority (async)
        priority_query = f"pam:escalation_priority({conflict_type}, Priority)"
        try:
            priority_result = await loop.run_in_executor(
                None, self._prolog.query_one, priority_query
            )
            priority = priority_result.get("Priority", 1) if priority_result else 1
        except Exception as e:
            logger.error(f"Prolog query failed for escalation priority: {e}")
            priority = 1

        return EscalationPath(
            steps=steps,
            final_authority=path_list[-1] if path_list else "U:Ox",
            priority=priority,
        )

    @traces(
        spec="001",
        story="US-003",
        reqs=[
            "REQ-001-US-003-023",
            "REQ-001-US-003-026",
            "REQ-001-US-003-030",
        ],
    )
    async def escalate(self, conflict: Conflict) -> EscalationResult:
        """Escalate conflict to appropriate authority (REQ-001-US-003-023 to REQ-001-US-003-027).

        Args:
            conflict: Conflict to escalate

        Returns:
            EscalationResult with escalation tracking info
        """
        escalation_id = f"esc-{uuid.uuid4().hex[:8]}"

        # Get escalation path
        path = await self.get_escalation_path(conflict)

        # Create escalation result
        timeout_at = _utc_now() + timedelta(seconds=self._default_timeout)

        result = EscalationResult(
            escalation_id=escalation_id,
            conflict=conflict,
            path=path,
            status=EscalationStatus.PENDING,
            timeout_at=timeout_at,
        )

        # Store escalation
        self._escalations[escalation_id] = result

        # Notify first authority in path (REQ-001-US-003-026)
        if path.steps:
            first_step = path.steps[0]
            first_step.notified_at = _utc_now()
            logger.info(
                f"Escalation {escalation_id} notified to {first_step.authority} "
                f"(priority: {path.priority})"
            )

        # Log to graph (REQ-001-US-003-030)
        if self._graph:
            await self._graph.log_escalation(result)

        return result

    @traces(
        spec="001",
        story="US-003",
        reqs=[
            "REQ-001-US-003-029",
            "REQ-001-US-005-021",
            "REQ-001-US-005-024",
        ],
    )
    async def resolve(self, escalation_id: str, resolution: Resolution) -> bool:
        """Resolve an escalation (REQ-001-US-003-029).

        Args:
            escalation_id: Escalation to resolve
            resolution: Resolution details

        Returns:
            True if resolved successfully
        """
        if escalation_id not in self._escalations:
            return False

        escalation = self._escalations[escalation_id]

        escalation.status = EscalationStatus.RESOLVED
        escalation.resolved_at = resolution.resolved_at
        escalation.resolution = resolution.resolution_text

        logger.info(
            f"Escalation {escalation_id} resolved by {resolution.resolved_by}: "
            f"{resolution.resolution_text}"
        )

        # Log to graph (REQ-001-US-003-030)
        if self._graph:
            await self._graph.log_resolution(escalation_id, resolution)

        return True

    @traces(
        spec="001",
        story="US-003",
        reqs=["REQ-001-US-003-025", "REQ-001-US-005-016"],
    )
    async def timeout_stale(self, max_age_seconds: int) -> list[str]:
        """Timeout stale escalations (REQ-001-US-003-025).

        Args:
            max_age_seconds: Maximum age before timeout

        Returns:
            List of timed out escalation IDs
        """
        timed_out = []
        for esc_id, escalation in self._escalations.items():
            # Check both PENDING and IN_PROGRESS escalations
            if escalation.status in (
                EscalationStatus.PENDING,
                EscalationStatus.IN_PROGRESS,
            ):
                if escalation.timeout_at and escalation.timeout_at < _utc_now():
                    escalation.status = EscalationStatus.TIMED_OUT
                    timed_out.append(esc_id)
                    logger.warning(f"Escalation {esc_id} timed out")

                    # Log to graph (REQ-001-US-003-030)
                    if self._graph:
                        await self._graph.log_timeout(esc_id)

        return timed_out

    @traces(spec="001", story="US-003", reqs=["REQ-001-US-003-024"])
    def get_escalation(self, escalation_id: str) -> EscalationResult | None:
        """Get escalation by ID (REQ-001-US-003-024).

        Args:
            escalation_id: Escalation ID

        Returns:
            EscalationResult if found
        """
        return self._escalations.get(escalation_id)

    @traces(
        spec="001",
        story="US-003",
        reqs=[
            "REQ-001-US-003-024",
        ],
    )
    def get_all_escalations(self) -> list[EscalationResult]:
        """Get all escalations.

        Returns:
            List of all escalation results
        """
        return list(self._escalations.values())


# ================================================================
# Advanced Escalation Integration (US-005)
# ================================================================


@runtime_checkable
class _EscalationLookup(Protocol):
    """Structural shape ``route_escalation`` actually needs.

    ``MultiHopEscalationManager`` is typed to accept a ``PAMEnforcer``, but
    ``route_escalation`` below calls ``.get_escalation()``, a method that
    lives on ``EscalationManager`` (this same module), not ``PAMEnforcer`` --
    ``PAMEnforcer`` has no escalation storage of its own. This is a
    pre-existing bug inherited from the source monorepo (same mismatch in
    ``orchestrata-kernel/pam_enforcer.py``); there are zero production
    callers of ``MultiHopEscalationManager`` (unwired), only tests, and the
    tests that exercise ``route_escalation`` pass an ``EscalationManager``
    despite the ``pam_enforcer`` parameter name. Typing the parameter to
    what the method genuinely requires documents the real contract without
    guessing at (or silently forcing) the missing production wiring
    decision.
    """

    def get_escalation(self, escalation_id: str) -> EscalationResult | None: ...


@traces(
    spec="001",
    story="US-003",
    reqs=[
        "REQ-001-US-003-023",
        "REQ-001-US-003-028",
        "REQ-001-US-005-014",
        "REQ-001-US-005-015",
    ],
)
class MultiHopEscalationManager:
    """Enhanced escalation manager with multi-hop support (US-005).

    Extends single-hop escalation with advanced multi-hop capabilities.
    Use this for advanced escalation scenarios. For basic escalation,
    use the original EscalationManager class.
    """

    def __init__(
        self,
        pam_enforcer: _EscalationLookup,
        prolog: IGovernancePrologBridge,
        redis_state: Any,
        graph_audit: Any | None = None,
    ) -> None:
        """Initialize multi-hop escalation manager.

        Args:
            pam_enforcer: PAM enforcer instance
            prolog: Prolog bridge
            redis_state: Redis state manager
            graph_audit: Optional Graph audit logger
        """
        self._pam = pam_enforcer
        self._prolog = prolog
        self._redis = redis_state
        self._graph = graph_audit

        # Create advanced escalation handler
        self._advanced_handler = AdvancedEscalationHandler(
            prolog=prolog, redis=redis_state, graph=graph_audit
        )

    @traces(spec="001", story="US-003", reqs=["REQ-001-US-003-023"])
    async def route_escalation(self, escalation_id: str, conflict: Any) -> str | None:
        """Route escalation (backward compatible single-hop).

        Args:
            escalation_id: Escalation ID
            conflict: Conflict to escalate

        Returns:
            Authority to route to, or None
        """
        # Use existing PAM enforcer logic
        escalation = self._pam.get_escalation(escalation_id)

        if not escalation:
            return None

        # Get escalation path from PAM
        path = escalation.path

        if path.steps:
            return path.steps[0].authority

        return None

    @traces(
        spec="001",
        story="US-003",
        reqs=["REQ-001-US-003-028", "REQ-001-US-005-015"],
    )
    async def multi_hop_escalate(
        self, conflict: AdvancedConflict, path: list[str]
    ) -> Any:
        """Execute multi-hop escalation (NEW - US-005).

        Args:
            conflict: Conflict to escalate
            path: Escalation path (list of volk aliases)

        Returns:
            EscalationResult with full hop history
        """
        # Convert path to starting volk (first in path)
        starting_volk = path[0] if path else "orchestrator"

        # Use advanced escalation handler
        result = await self._advanced_handler.escalate(
            conflict=conflict, starting_volk=starting_volk
        )

        return result