orchestrata_ultimate.agents.escalation

Reference › orchestrata_ultimate › agents › escalation

escalation moves an issue up the agent hierarchy while preserving its context and tracking its resolution (specification 005, user stories US-002 and US-006). AgentEscalation is built over an AgentHierarchy (hierarchy), an optional AgentCommunication (communication) and an optional MultiagentPrologBridge.

get_escalation_path returns the ordered chain from an agent to the root. escalate records an EscalationResult (id, source, target, issue, path, status, context and duration) and sends the issue to the next level. resolve marks it resolved with a resolution text. auto_escalate_if_needed escalates only when an action is outside the agent's scope. Lookup helpers are get_escalation, get_pending_escalations and get_agent_escalations; total_escalations and pending_count are properties. The module is not imported by other reference modules; it is exercised by tests.

At a glance

Class diagram

Dependency diagram

API

Classes

EscalationResult (dataclass)

Decorators: dataclass, traces(spec='005', story='US-006', reqs=['REQ-005-US-006-001']) · Line 35

Result of an escalation.

AgentEscalation

Decorators: traces(spec='005', story='US-002', reqs=['REQ-005-FR7-001', 'REQ-005-US-002-001', 'REQ-005-US-006-001']) · Line 74

Manages escalation of issues through the agent hierarchy.

Source

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

# mypy: ignore-errors
"""Agent-level escalation management.

Handles escalation of issues up the agent hierarchy with
context preservation, resolution tracking, and auto-escalation.
"""

# Spec: 005 - Multi-Agent System

from __future__ import annotations

import logging
import time
import uuid
from dataclasses import dataclass, field
from datetime import UTC, datetime
from typing import TYPE_CHECKING, Any

from orchestrata_ultimate.agents.hierarchy import AgentHierarchy
from orchestrata_ultimate.core.base.traces import traces

if TYPE_CHECKING:
    from orchestrata_ultimate.agents.communication import AgentCommunication
    from orchestrata_ultimate.agents.prolog_bridge import MultiagentPrologBridge

logger = logging.getLogger(__name__)


@dataclass
@traces(
    spec="005",
    story="US-006",
    reqs=["REQ-005-US-006-001"],
)
class EscalationResult:
    """Result of an escalation."""

    escalation_id: str
    source_agent: str
    target_agent: str
    issue: str
    path: list[str]
    status: str = "escalated"  # "escalated", "resolved", "rejected"
    resolution: str | None = None
    resolved_at: datetime | None = None
    created_at: datetime = field(default_factory=lambda: datetime.now(UTC))
    context: dict[str, Any] = field(default_factory=dict)
    duration_ms: float = 0.0

    def to_dict(self) -> dict[str, Any]:
        return {
            "escalation_id": self.escalation_id,
            "source_agent": self.source_agent,
            "target_agent": self.target_agent,
            "issue": self.issue,
            "path": self.path,
            "status": self.status,
            "resolution": self.resolution,
            "created_at": self.created_at.isoformat(),
            "resolved_at": self.resolved_at.isoformat() if self.resolved_at else None,
            "duration_ms": self.duration_ms,
        }


@traces(
    spec="005",
    story="US-002",
    reqs=[
        "REQ-005-FR7-001",
        "REQ-005-US-002-001",
        "REQ-005-US-006-001",
    ],
)
class AgentEscalation:
    """Manages escalation of issues through the agent hierarchy.

    Escalation follows the hierarchy path upward. Context is
    preserved during escalation (Article III). Escalation cannot
    be suppressed (Article III).
    """

    def __init__(
        self,
        hierarchy: AgentHierarchy,
        communication: AgentCommunication | None = None,
        prolog_bridge: MultiagentPrologBridge | None = None,
    ) -> None:
        self._hierarchy = hierarchy
        self._communication = communication
        self._prolog = prolog_bridge
        self._escalations: dict[str, EscalationResult] = {}

    @traces(
        spec="005",
        story="US-002",
        reqs=[
            "REQ-005-AC2-001",
            "REQ-005-FR2-001",
            "REQ-005-NFR4-001",
            "REQ-005-US-002-001",
        ],
    )
    def get_escalation_path(self, agent_id: str) -> list[str]:
        """Get the ordered escalation path from agent to root."""
        return self._hierarchy.get_ancestors(agent_id)

    @traces(
        spec="005",
        story="US-002",
        reqs=[
            "REQ-005-AC4-001",
            "REQ-005-FR7-001",
            "REQ-005-US-002-001",
        ],
    )
    def escalate(
        self,
        agent_id: str,
        issue: str,
        context: dict[str, Any] | None = None,
    ) -> EscalationResult:
        """Escalate an issue from the given agent up the hierarchy."""
        start = time.perf_counter()
        path = self.get_escalation_path(agent_id)

        if not path:
            # Root agent — cannot escalate further
            result = EscalationResult(
                escalation_id=str(uuid.uuid4()),
                source_agent=agent_id,
                target_agent=agent_id,
                issue=issue,
                path=[],
                status="rejected",
                context=context or {},
            )
            result.duration_ms = (time.perf_counter() - start) * 1000
            self._escalations[result.escalation_id] = result
            return result

        target = path[0]  # Immediate parent

        result = EscalationResult(
            escalation_id=str(uuid.uuid4()),
            source_agent=agent_id,
            target_agent=target,
            issue=issue,
            path=path,
            status="escalated",
            context=context or {},
        )
        result.duration_ms = (time.perf_counter() - start) * 1000

        # Send via communication if available
        if self._communication is not None:
            try:
                self._communication.send_to_parent(
                    agent_id,
                    {
                        "type": "escalation",
                        "escalation_id": result.escalation_id,
                        "issue": issue,
                        "context": context or {},
                    },
                )
            except Exception as exc:
                logger.warning(f"Failed to send escalation message: {exc}")

        self._escalations[result.escalation_id] = result
        return result

    @traces(
        spec="005",
        story="US-002",
        reqs=["REQ-005-AC4-001"],
    )
    def resolve(self, escalation_id: str, resolution: str) -> EscalationResult | None:
        """Resolve an escalation."""
        esc = self._escalations.get(escalation_id)
        if esc is None:
            return None

        esc.status = "resolved"
        esc.resolution = resolution
        esc.resolved_at = datetime.now(UTC)

        # Notify source agent via communication
        if self._communication is not None:
            try:
                self._communication.send_to_child(
                    esc.target_agent,
                    esc.source_agent,
                    {
                        "type": "escalation_resolved",
                        "escalation_id": escalation_id,
                        "resolution": resolution,
                    },
                )
            except Exception as exc:
                logger.warning(
                    "Failed to notify source agent '%s' of escalation resolution '%s': %s",
                    esc.source_agent,
                    escalation_id,
                    exc,
                    exc_info=True,
                )

        return esc

    @traces(
        spec="005",
        story="US-006",
        reqs=["REQ-005-FR7-001", "REQ-005-US-006-001"],
    )
    def auto_escalate_if_needed(
        self,
        agent_id: str,
        action: str,
        context: dict[str, Any] | None = None,
    ) -> EscalationResult | None:
        """Automatically escalate if the action is outside the agent's scope."""
        # Check via Prolog if available
        if self._prolog is not None:
            try:
                can_do = self._prolog.can_start_volk(agent_id)
                if can_do:
                    return None  # No escalation needed
            except Exception as exc:
                logger.warning(
                    "Prolog can_start_volk check failed for agent '%s' during auto-escalation: %s",
                    agent_id,
                    exc,
                    exc_info=True,
                )

        # Check if action requires escalation based on naming convention
        escalation_actions = {
            "terminate",
            "force_stop",
            "modify_constraints",
            "cross_scope",
        }
        if action in escalation_actions:
            return self.escalate(
                agent_id,
                f"Auto-escalation: {action} requires approval",
                context=context,
            )

        return None

    def get_escalation(self, escalation_id: str) -> EscalationResult | None:
        """Get an escalation by ID."""
        return self._escalations.get(escalation_id)

    @traces(spec="005", story="US-004", reqs=["REQ-005-AC6-001", "REQ-005-FR8-001"])
    def get_pending_escalations(self) -> list[EscalationResult]:
        """Get all unresolved escalations."""
        return [e for e in self._escalations.values() if e.status == "escalated"]

    def get_agent_escalations(self, agent_id: str) -> list[EscalationResult]:
        """Get all escalations from a specific agent."""
        return [e for e in self._escalations.values() if e.source_agent == agent_id]

    @property
    @traces(
        spec="005",
        story="US-004",
        reqs=["REQ-005-AC6-001", "REQ-005-FR8-001"],
    )
    def total_escalations(self) -> int:
        return len(self._escalations)

    @property
    @traces(
        spec="005",
        story="US-004",
        reqs=["REQ-005-AC6-001", "REQ-005-FR8-001"],
    )
    def pending_count(self) -> int:
        return len(self.get_pending_escalations())