orchestrata_ultimate.verification.escalation_handler

Reference › orchestrata_ultimate › verification › escalation_handler

EscalationHandler decides when verification gives up and hands a session to a human. It is used by the orchestrator's verification_gate and traces Spec 035 US-008. should_escalate returns true once a VerificationSession's attempt_count reaches the handler's max_attempts (default 3). escalate_to_human wraps the outcome in an EscalationMessage, and build_escalation_report produces the detail: constraints total, evaluated, satisfied and failed, attempts, the current checkpoint, and per-constraint failure status and details. It reads only the session object; it touches no external store. Session and message types come from types.

At a glance

Class diagram

Dependency diagram

API

Classes

EscalationHandler

Decorators: traces(spec='035', story='US-008', reqs=['REQ-035-US-008-005', 'REQ-035-US-008-006', 'REQ-035-US-008-014', 'REQ-035-US-… · Line 28

Handles escalation when verification fails repeatedly.

Source

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

"""Escalation handler — max attempts to human."""

# Spec: 035 - Verification Agent

import logging
from typing import Any

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.verification.types import (
    EscalationMessage,
    SessionStatus,
    VerificationSession,
)

logger = logging.getLogger(__name__)


@traces(
    spec="035",
    story="US-008",
    reqs=[
        "REQ-035-US-008-005",
        "REQ-035-US-008-006",
        "REQ-035-US-008-014",
        "REQ-035-US-008-021",
    ],
)
class EscalationHandler:
    """Handles escalation when verification fails repeatedly."""

    def __init__(self, max_attempts: int = 3) -> None:
        self._max_attempts = max_attempts

    @traces(
        spec="035", story="US-008", reqs=["REQ-035-US-008-006", "REQ-035-US-008-021"]
    )
    def should_escalate(self, session: VerificationSession) -> bool:
        """Check if a session requires escalation."""
        return session.attempt_count >= self._max_attempts

    @traces(spec="035", story="US-008", reqs=["REQ-035-US-008-005"])
    def escalate_to_human(self, session: VerificationSession) -> EscalationMessage:
        """Create an escalation message for human review."""
        failed = [cid for cid, r in session.results.items() if not r.is_satisfied]
        session.status = SessionStatus.ESCALATED

        return EscalationMessage(
            session_id=session.id,
            reason=f"Verification failed after {session.attempt_count} attempts",
            failed_constraints=failed,
            attempt_count=session.attempt_count,
            report=self.build_escalation_report(session),
        )

    @traces(spec="035", story="US-008", reqs=["REQ-035-US-008-005"])
    def build_escalation_report(self, session: VerificationSession) -> dict[str, Any]:
        """Build a detailed escalation report."""
        satisfied = sum(1 for r in session.results.values() if r.is_satisfied)
        total = len(session.results)

        return {
            "session_id": session.id,
            "total_constraints": len(session.constraints),
            "evaluated": total,
            "satisfied": satisfied,
            "failed": total - satisfied,
            "attempts": session.attempt_count,
            "current_checkpoint": session.current_checkpoint.value,
            "failed_details": [
                {"constraint_id": cid, "status": r.status.value, "details": r.details}
                for cid, r in session.results.items()
                if not r.is_satisfied
            ],
        }