orchestrata_ultimate.governance.escalation_handler.handler

Reference › orchestrata_ultimate › governance › escalation_handler › handler

Multi-hop escalation, the part of governance that carries an unresolved conflict up the chain of authority, SPEC-001 story US-005 (REQ-001-US-005-001 to -017). AdvancedEscalationHandler takes a Conflict and a starting volk, asks Prolog for the escalation path of the affected component, then sends an escalation packet to each authority in turn until one accepts or the chain reaches a terminal authority. The kernel's MultiHopEscalationManager delegates to it, and orchestrator.graph_audit imports the package. The SLA tracker in sla imports this module, and the data types are in models.

escalate makes an escalation id, calls get_escalation_path (which queries pam:multi_hop_escalation/2 and falls back to the single authority orchestrator), runs execute_multi_hop, builds an EscalationResult from the hops and the elapsed time, stores it in Redis (_store_escalation) and writes it to the graph audit trail. execute_multi_hop walks the path creating an EscalationHop with a deadline (DEFAULT_HOP_TIMEOUT_SECONDS, 300 seconds) for each volk alias; _execute_hop retries up to the configured maximum, and _attempt_hop_once marks a hop timed out when its deadline has passed (calling handle_escalation_timeout), otherwise builds a packet and dispatches it. The walk stops at the first hop that succeeds or that targets a terminal authority (_is_terminal_authority); a failed hop moves on to the next.

The message that moves between packages: _build_escalation_packet creates a CXPPacket from governance to the hop's volk with subject escalation, high priority, and a payload naming the conflict id and type, component, volks, description, hop index and retry count. _dispatch_packet hands it to a route callback supplied by the caller, which is how the packet reaches the postman; with no callback it logs a warning and reports failure. track_escalation_state keeps hop state in Redis, handle_escalation_timeout records the timeout in Redis and the graph, and override_escalation_path lets an operator replace the path.

At a glance

Class diagram

Dependency diagram

API

Classes

_AsyncClientRedisStateManager (protocol)

Bases: IRedisStateManager, Protocol · Decorators: runtime_checkable · Line 49

IRedisStateManager plus the async_client property this handler needs.

AdvancedEscalationHandler

Decorators: traces(spec='001', story='US-005', reqs=['REQ-001-US-005-011', 'REQ-001-US-005-014']) · Line 70

Handles advanced multi-hop escalation logic (US-005).

Functions

Source

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

# mypy: ignore-errors
"""Advanced multi-hop escalation handler (US-005)."""

import asyncio
import json
import logging
import time
import uuid
from collections.abc import Callable
from datetime import timedelta
from typing import Any, Protocol, runtime_checkable

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.core.packet import CXPPacket, Priority
from orchestrata_ultimate.core.protocols import (
    IAuditTrail,
    IGovernancePrologBridge,
    IRedisStateManager,
)
from orchestrata_ultimate.governance.constants import (
    DEFAULT_ESCALATION_MAX_RETRIES,
    DEFAULT_HOP_TIMEOUT_SECONDS,
)
from orchestrata_ultimate.governance.escalation_handler.models import (
    Conflict,
    EscalationHop,
    EscalationResult,
    EscalationStatus,
    _utc_now,
)

logger = logging.getLogger(__name__)


def _quote_atom(value: str) -> str:
    """Escape a string for safe use inside a single-quoted Prolog atom.

    Without this, an unquoted volk alias like ``U:Cd`` parses as the
    module-qualification OPERATOR (``:``) applied to two unbound
    variables (``U`` and ``Cd``, both uppercase) rather than the intended
    atom -- a silent wrong-contract bug distinct from, but adjacent to,
    F4's nonexistent-predicate defect (both query sites this backs were
    touched to fix F4).
    """
    return "'" + value.replace("\\", "\\\\").replace("'", "\\'") + "'"


@runtime_checkable
class _AsyncClientRedisStateManager(IRedisStateManager, Protocol):
    """``IRedisStateManager`` plus the ``async_client`` property this
    handler needs.

    ``IRedisStateManager`` (``core/protocols/decoupling.py``) is the shared
    cycle-breaking decoupling interface and deliberately doesn't declare
    ``async_client`` (most consumers only need the sync surface). This
    handler needs the async Redis client (escalation hops are async),
    so it extends the shared Protocol locally rather than widening it
    project-wide (precedent: escalation_handler/resolver.py).
    """

    @property
    def async_client(self) -> Any: ...


@traces(
    spec="001",
    story="US-005",
    reqs=["REQ-001-US-005-011", "REQ-001-US-005-014"],
)
class AdvancedEscalationHandler:
    """Handles advanced multi-hop escalation logic (US-005)."""

    def __init__(
        self,
        prolog: IGovernancePrologBridge,
        redis: _AsyncClientRedisStateManager,
        graph: IAuditTrail | None = None,
        default_hop_timeout: int = DEFAULT_HOP_TIMEOUT_SECONDS,
        max_retries: int = DEFAULT_ESCALATION_MAX_RETRIES,
        route_callback: Callable[[CXPPacket], bool] | None = None,
    ) -> None:
        """Initialize escalation handler.

        Args:
            prolog: Prolog bridge for querying rules
            redis: Redis state manager
            graph: Optional graph audit logger
            default_hop_timeout: Default timeout per hop in seconds
            max_retries: Maximum retries per hop
            route_callback: Optional callback to dispatch CXP
                packets. Returns True on success. Falls back
                to logging when None.
        """
        self._prolog = prolog
        self._redis_manager = redis
        self._redis = redis.async_client  # Use async Redis client
        self._graph = graph
        self._default_hop_timeout = default_hop_timeout
        self._max_retries = max_retries
        self._route_callback = route_callback

        # In-memory cache (backed by Redis)
        self._escalations: dict[str, EscalationResult] = {}

    @traces(spec="001", reqs=["REQ-001-US-005-003"])
    async def escalate(
        self, conflict: Conflict, starting_volk: str
    ) -> EscalationResult:
        """Execute multi-hop escalation (REQ-001-US-005-001).

        Args:
            conflict: Conflict to escalate
            starting_volk: Volk that initiated escalation

        Returns:
            EscalationResult with full escalation history
        """
        escalation_id = f"esc-{uuid.uuid4().hex[:8]}"
        start_time = time.time()

        path = await self.get_escalation_path(conflict.component)

        logger.info(
            f"Starting escalation {escalation_id} for conflict {conflict.conflict_id} "
            f"from {starting_volk} through path: {path}"
        )

        hops = await self.execute_multi_hop(path, conflict)
        total_time_ms = int((time.time() - start_time) * 1000)

        result = self._build_escalation_result(
            escalation_id, conflict, hops, total_time_ms
        )

        self._escalations[escalation_id] = result
        await self._store_escalation(result)

        await self._audit_escalation_to_graph(
            escalation_id, conflict, path, result, starting_volk, total_time_ms, hops
        )

        return result

    @traces(spec="001", story="US-005", reqs=["REQ-001-US-005-021"])
    def _build_escalation_result(
        self,
        escalation_id: str,
        conflict: Conflict,
        hops: list,
        total_time_ms: int,
    ) -> EscalationResult:
        """Determine resolution state and construct the EscalationResult."""
        resolved = any(hop.status == EscalationStatus.completed for hop in hops)
        resolution = None
        resolved_at = None

        if resolved:
            completed_hop = next(
                hop for hop in hops if hop.status == EscalationStatus.completed
            )
            resolution = completed_hop.response
            resolved_at = _utc_now()

        return EscalationResult(
            escalation_id=escalation_id,
            conflict=conflict,
            hops=hops,
            resolved=resolved,
            resolution=resolution,
            total_time_ms=total_time_ms,
            resolved_at=resolved_at,
        )

    @traces(spec="001", story="US-008", reqs=["REQ-001-US-008-011"])
    async def _audit_escalation_to_graph(
        self,
        escalation_id: str,
        conflict: Conflict,
        path: list,
        result: EscalationResult,
        starting_volk: str,
        total_time_ms: int,
        hops: list,
    ) -> None:
        """Log the completed escalation to graph audit trail (REQ-001-US-005-020)."""
        if not self._graph:
            return

        from orchestrata_ultimate.orchestrator.graph_audit import (  # noqa: PLC0415
            AuditEvent,
            AuditEventType,
        )

        event = AuditEvent(
            event_type=AuditEventType.ESCALATION,
            session_id=escalation_id,
            volk_id=starting_volk,
            timestamp=_utc_now(),
            data={
                "escalation_id": escalation_id,
                "conflict_id": conflict.conflict_id,
                "path": path,
                "resolved": result.resolved,
                "total_time_ms": total_time_ms,
                "hop_count": len(hops),
            },
        )
        await self._graph.log_event(event)

    @traces(spec="001", story="US-005", reqs=["REQ-001-US-005-014"])
    async def execute_multi_hop(
        self, path: list[str], conflict: Conflict
    ) -> list[EscalationHop]:
        """Execute multi-hop escalation sequence (REQ-001-US-005-001, REQ-001-US-005-002).

        Args:
            path: List of volk aliases in escalation path
            conflict: Conflict being escalated

        Returns:
            List of escalation hops with statuses
        """
        hops: list[EscalationHop] = []

        for hop_index, volk_alias in enumerate(path):
            # Check if terminal authority (REQ-001-US-005-008)
            is_terminal = await self._is_terminal_authority(volk_alias)

            # Create hop
            hop = EscalationHop(
                hop_index=hop_index,
                volk_alias=volk_alias,
                status=EscalationStatus.pending,
                timestamp=_utc_now(),
                timeout_at=_utc_now() + timedelta(seconds=self._default_hop_timeout),
            )

            # Execute hop with retry logic (REQ-001-US-005-010)
            success = await self._execute_hop(hop, conflict)

            # Track state (REQ-001-US-005-002)
            await self.track_escalation_state(
                f"esc-{conflict.conflict_id}", hop_index, hop.status
            )

            # Log hop (REQ-001-US-005-005)
            await self._log_escalation_hop(conflict.conflict_id, hop)

            hops.append(hop)

            # If resolved or terminal authority reached, stop
            if success or is_terminal:
                hop.status = EscalationStatus.completed
                hop.response = f"Resolved by {volk_alias}"
                break

            # If hop failed after retries, continue to next
            if hop.status == EscalationStatus.failed:
                logger.warning(
                    f"Hop {hop_index} to {volk_alias} failed after {hop.retry_count} retries"
                )
                continue

        return hops

    async def _execute_hop(self, hop: EscalationHop, conflict: Conflict) -> bool:
        """Execute a single escalation hop with retry (REQ-001-US-005-010).

        Args:
            hop: Hop to execute
            conflict: Conflict being escalated

        Returns:
            True if hop succeeded, False otherwise
        """
        hop.status = EscalationStatus.in_progress

        for retry in range(self._max_retries):
            hop.retry_count = retry
            succeeded, done = await self._attempt_hop_once(hop, conflict, retry)
            if done:
                return succeeded

        # All retries exhausted
        hop.status = EscalationStatus.failed  # pragma: no cover
        return False  # pragma: no cover

    @traces(spec="001", story="US-005", reqs=["REQ-001-US-005-017"])
    def _build_escalation_packet(
        self, hop: EscalationHop, conflict: Conflict
    ) -> CXPPacket:
        """Build a CXP packet for escalation routing.

        Args:
            hop: Current escalation hop
            conflict: Conflict being escalated

        Returns:
            CXPPacket addressed to the target volk
        """
        payload = {
            "conflict_id": conflict.conflict_id,
            "conflict_type": conflict.conflict_type.value,
            "component": conflict.component,
            "volks": conflict.volks,
            "description": conflict.description,
            "hop_index": hop.hop_index,
            "retry_count": hop.retry_count,
        }
        return CXPPacket(
            from_agent="governance",
            to_agent=hop.volk_alias,
            subject="escalation",
            payload=payload,
            priority=Priority.HIGH,
        )

    def _dispatch_packet(self, packet: CXPPacket) -> bool:
        """Dispatch a CXP packet via the route callback.

        Falls back to logging when no callback is configured.

        Args:
            packet: CXP packet to dispatch

        Returns:
            True if dispatched successfully
        """
        if self._route_callback is None:
            logger.warning(
                "No route_callback configured; cannot dispatch packet %s to %s",
                packet.message_id,
                packet.to_agent,
            )
            return False
        return self._route_callback(packet)

    @traces(spec="001", story="US-005", reqs=["REQ-001-US-005-016"])
    async def _attempt_hop_once(
        self, hop: EscalationHop, conflict: Conflict, retry: int
    ) -> tuple:
        """Perform one attempt of a hop, handling timeout and routing.

        Returns:
            (succeeded, done) where done=True means stop retrying.
        """
        try:
            if hop.timeout_at and _utc_now() > hop.timeout_at:
                hop.status = EscalationStatus.timeout
                await self.handle_escalation_timeout(
                    f"esc-{conflict.conflict_id}",
                    hop.hop_index,
                )
                return False, True

            packet = self._build_escalation_packet(hop, conflict)
            dispatched = self._dispatch_packet(packet)

            if dispatched:
                hop.status = EscalationStatus.completed
                hop.response = f"Routed to {hop.volk_alias}"
                return True, True

            # Dispatch failed — retry or give up
            if retry < self._max_retries - 1:
                await asyncio.sleep(0.01)
                return False, False

            hop.status = EscalationStatus.failed
            return False, True

        except Exception as e:
            logger.error(
                "Error in hop %d retry %d: %s",
                hop.hop_index,
                retry,
                e,
            )
            if retry < self._max_retries - 1:
                await asyncio.sleep(0.01)
                return False, False
            else:  # pragma: no cover
                hop.status = EscalationStatus.failed  # pragma: no cover
                return False, True  # pragma: no cover

    @traces(
        spec="001",
        story="US-003",
        reqs=["REQ-001-US-003-024"],
    )
    async def track_escalation_state(
        self, escalation_id: str, hop_index: int, status: EscalationStatus
    ) -> None:
        """Track escalation state at each hop (REQ-001-US-005-002).

        Args:
            escalation_id: Escalation ID
            hop_index: Index of current hop
            status: Status of hop
        """
        key = f"escalation:{escalation_id}"

        state = {
            "current_hop": hop_index,
            "status": status.value,
            "updated_at": _utc_now().isoformat(),
        }

        try:
            await self._redis.set(key, json.dumps(state))
        except Exception as e:
            logger.error(f"Failed to track escalation state for {escalation_id}: {e}")
            raise

        logger.debug(
            f"Escalation {escalation_id} hop {hop_index} state: {status.value}"
        )

    @traces(
        spec="001",
        story="US-005",
        reqs=["REQ-001-US-005-016"],
    )
    async def handle_escalation_timeout(
        self, escalation_id: str, hop_index: int
    ) -> None:
        """Handle escalation timeout at specific hop (REQ-001-US-005-003).

        Args:
            escalation_id: Escalation ID
            hop_index: Index of hop that timed out
        """
        logger.warning(f"Escalation {escalation_id} hop {hop_index} timed out")

        try:
            # Update state
            key = f"escalation:{escalation_id}"
            state = await self._redis.get(key)

            if state:
                state_dict = json.loads(state)
                state_dict["timeout_hop"] = hop_index
                state_dict["timeout_at"] = _utc_now().isoformat()
            else:
                # Create new timeout record if no existing state
                state_dict = {
                    "escalation_id": escalation_id,
                    "timeout_hop": hop_index,
                    "timeout_at": _utc_now().isoformat(),
                    "status": "timed_out",
                }

            await self._redis.set(key, json.dumps(state_dict))
        except Exception as e:
            logger.error(f"Failed to update timeout state for {escalation_id}: {e}")
            raise

        # Log to graph (graph.log_event may need await if it's async)
        if self._graph:
            from orchestrata_ultimate.orchestrator.graph_audit import (  # noqa: PLC0415
                AuditEvent,
                AuditEventType,
            )

            event = AuditEvent(
                event_type=AuditEventType.ESCALATION,
                session_id=escalation_id,
                volk_id="system",
                timestamp=_utc_now(),
                data={
                    "escalation_id": escalation_id,
                    "hop_index": hop_index,
                    "action": "timeout",
                },
            )
            await self._graph.log_event(event)

    async def override_escalation_path(
        self, escalation_id: str, new_path: list[str]
    ) -> None:
        """Emergency override of escalation path (REQ-001-US-005-004).

        Args:
            escalation_id: Escalation to override
            new_path: New escalation path
        """
        logger.warning(
            f"Emergency override of escalation {escalation_id} to path: {new_path}"
        )

        try:
            key = f"escalation_override:{escalation_id}"
            await self._redis.set(key, json.dumps(new_path))
        except Exception as e:
            logger.error(f"Failed to override escalation path for {escalation_id}: {e}")
            raise

        # Log override
        if self._graph:
            from orchestrata_ultimate.orchestrator.graph_audit import (  # noqa: PLC0415
                AuditEvent,
                AuditEventType,
            )

            event = AuditEvent(
                event_type=AuditEventType.ESCALATION,
                session_id=escalation_id,
                volk_id="system",
                timestamp=_utc_now(),
                data={
                    "escalation_id": escalation_id,
                    "new_path": new_path,
                    "action": "override",
                },
            )
            await self._graph.log_event(event)

    @traces(spec="001", story="US-003", reqs=["REQ-001-US-003-022"])
    async def get_escalation_path(self, component: str) -> list[str]:
        """Query Prolog for escalation path (REQ-001-US-005-001).

        Args:
            component: Component being escalated

        Returns:
            List of volk aliases in escalation order
        """
        # Query Prolog for multi-hop escalation path (pam_rules.pl -- F4:
        # this predicate did not exist before this fix; every call errored
        # into the "default to orchestrator" fallback below regardless of
        # the real PAM ownership/authority chain). Module-qualified + the
        # component atom quoted (component names are plain lowercase
        # words today, but quoting makes the query correct regardless).
        query = f"pam:multi_hop_escalation({_quote_atom(component)}, Path)"
        loop = asyncio.get_event_loop()
        result = await loop.run_in_executor(None, self._prolog.query_one, query)

        if result and "Path" in result:
            path = result["Path"]
            # Convert Prolog list to Python list
            if isinstance(path, list):
                return path
            return [path]

        # Default to orchestrator
        return ["orchestrator"]

    @traces(spec="001", story="US-005", reqs=["REQ-001-US-005-014"])
    async def _is_terminal_authority(self, volk_alias: str) -> bool:
        """Check if volk is terminal authority (REQ-001-US-005-008).

        Args:
            volk_alias: Volk alias to check

        Returns:
            True if terminal authority
        """
        # F4: terminal_authority/1 did not exist before this fix; module-
        # qualified + quoted for the same reason as get_escalation_path
        # above -- a bare volk alias like 'U:Cd' is NOT a valid unquoted
        # Prolog atom (':' is the module-qualification operator, so it
        # would parse as two unbound variables, not the intended atom).
        query = f"pam:terminal_authority({_quote_atom(volk_alias)})"
        loop = asyncio.get_event_loop()
        return await loop.run_in_executor(None, self._prolog.query_bool, query)

    @traces(spec="001", story="US-005", reqs=["REQ-001-US-005-019"])
    async def _log_escalation_hop(self, conflict_id: str, hop: EscalationHop) -> None:
        """Log escalation hop details (REQ-001-US-005-005).

        Args:
            conflict_id: Conflict ID
            hop: Hop to log
        """
        hop_key = f"escalation_hops:esc-{conflict_id}"

        hop_data = {
            "hop_index": hop.hop_index,
            "volk_alias": hop.volk_alias,
            "status": hop.status.value,
            "timestamp": hop.timestamp.isoformat(),
            "response": hop.response,
            "retry_count": hop.retry_count,
        }

        try:
            # Append to hop history in Redis
            await self._redis.lpush(hop_key, json.dumps(hop_data))
        except Exception as e:
            logger.error(f"Failed to log escalation hop for {conflict_id}: {e}")
            raise

        logger.info(
            f"Logged escalation hop {hop.hop_index} to {hop.volk_alias}: {hop.status.value}"
        )

    @traces(spec="001", story="US-005", reqs=["REQ-001-US-005-028"])
    async def _store_escalation(self, result: EscalationResult) -> None:
        """Store escalation result in Redis.

        Args:
            result: Escalation result to store
        """
        key = f"escalation:{result.escalation_id}"

        # Determine current hop and status
        current_hop = len(result.hops) - 1 if result.hops else 0
        if result.resolved:
            status = "completed"
        elif any(hop.status == EscalationStatus.failed for hop in result.hops):
            status = "failed"
        elif any(hop.status == EscalationStatus.timeout for hop in result.hops):
            status = "timeout"
        else:
            status = "in_progress"

        data = {
            "conflict_id": result.conflict.conflict_id,
            "resolved": result.resolved,
            "resolution": result.resolution,
            "total_time_ms": result.total_time_ms,
            "hop_count": len(result.hops),
            "current_hop": current_hop,
            "status": status,
            "created_at": result.created_at.isoformat(),
            "updated_at": _utc_now().isoformat(),
            "resolved_at": (
                result.resolved_at.isoformat() if result.resolved_at else None
            ),
        }

        try:
            await self._redis.set(key, json.dumps(data))
        except Exception as e:
            logger.error(f"Failed to store escalation {result.escalation_id}: {e}")
            raise