orchestrata_ultimate.governance.escalation_handler.resolver

Reference › orchestrata_ultimate › governance › escalation_handler › resolver

Conflict detection and resolution, SPEC-001 story US-005 (REQ-001-US-005-001, -002, -011 to -021). ConflictResolver finds out whether volks are in conflict over a component and records how a conflict is settled. The package initialiser of the escalation_handler subpackage imports it. When a conflict is not settled here, handler escalates it. Value types are in models.

detect_conflict(action, component, volks) asks Prolog. If more than one volk is given, it checks pam:pam_owns for each and, when more than one owns the component, returns an overlapping_authority conflict naming them. Otherwise it checks pam:cross_boundary_allowed for each volk and returns a boundary_violation conflict for the first that is not allowed, or None. create_resolution_workflow makes a workflow id, assign_to_authority records the assignee and track_resolution_status moves a workflow through a ResolutionStatus. apply_resolution_strategy(conflict, strategy) creates a workflow and a resolution id, applies the strategy, logs the decision and, when a graph repository is present, writes a :Resolution node and an escalation audit event. The three strategies are simple: _apply_vote_strategy and _apply_priority_strategy pick the first volk of the conflict (a code comment says the vote is a simulation), and _apply_delegation_strategy delegates to the orchestrator; each returns a text description, with no unwinnable case other than an empty volk list.

At a glance

Class diagram

Dependency diagram

API

Classes

_AsyncClientRedisStateManager (protocol)

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

IRedisStateManager plus the async_client property this resolver needs.

_RepoBackedAuditTrail (protocol)

Bases: IAuditTrail, Protocol · Line 48

IAuditTrail plus the _repo the resolver writes a Resolution node through directly.

ConflictResolver

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

Handles conflict detection and resolution (REQ-001-US-005-011 to REQ-001-US-005-020).

Source

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

# mypy: ignore-errors
"""Conflict detection and resolution (REQ-001-US-005-011 to REQ-001-US-005-020)."""

import asyncio
import json
import logging
import uuid
from typing import TYPE_CHECKING, Any, Protocol, runtime_checkable

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.core.protocols import (
    IAuditTrail,
    IGovernancePrologBridge,
    IRedisStateManager,
)
from orchestrata_ultimate.governance.escalation_handler.models import (
    Conflict,
    ConflictType,
    Resolution,
    ResolutionStatus,
    ResolutionStrategy,
    _utc_now,
)

if TYPE_CHECKING:
    from orchestrata_ultimate.core.graph import AsyncGraphRepository

logger = logging.getLogger(__name__)


@runtime_checkable
class _AsyncClientRedisStateManager(IRedisStateManager, Protocol):
    """``IRedisStateManager`` plus the ``async_client`` property this
    resolver 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
    resolver is one of the few that needs the async Redis client
    (``detect_conflict``/etc. below are async), so it extends the shared
    Protocol locally rather than widening it project-wide.
    """

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


class _RepoBackedAuditTrail(IAuditTrail, Protocol):
    """``IAuditTrail`` plus the ``_repo`` the resolver writes a Resolution
    node through directly.

    ``IAuditTrail`` only exposes the queue-based ``log_*`` surface (breaks
    the governance/orchestrator cycle); ``_apply_priority_strategy``'s
    Resolution-node write below is richer than ``log_resolution``'s enqueued
    event shape (it needs resolution_id/workflow_id/conflict_id/strategy),
    so it goes through the concrete ``GraphAuditTrail``'s repository
    directly. Extends the shared Protocol locally rather than widening it.
    """

    _repo: AsyncGraphRepository | None


@traces(spec="001", story="US-005", reqs=["REQ-001-US-005-001", "REQ-001-US-005-021"])
class ConflictResolver:
    """Handles conflict detection and resolution (REQ-001-US-005-011 to REQ-001-US-005-020)."""

    def __init__(
        self,
        prolog: IGovernancePrologBridge,
        redis: _AsyncClientRedisStateManager,
        graph: _RepoBackedAuditTrail | None = None,
    ) -> None:
        """Initialize conflict resolver.

        Args:
            prolog: Prolog bridge
            redis: Redis state manager
            graph: Optional graph audit logger
        """
        self._prolog = prolog
        self._redis_manager = redis
        self._redis = redis.async_client  # Use async Redis client
        self._graph = graph

        # Workflow tracking
        self._workflows: dict[str, dict[str, Any]] = {}

    @traces(
        spec="001", story="US-005", reqs=["REQ-001-US-005-001", "REQ-001-US-005-002"]
    )
    async def detect_conflict(
        self, action: str, component: str, volks: list[str]
    ) -> Conflict | None:
        """Detect if action represents a conflict (REQ-001-US-005-011, REQ-001-US-005-012).

        Args:
            action: Action being attempted
            component: Component involved
            volks: Volks involved

        Returns:
            Conflict if detected, None otherwise
        """
        # Detect overlapping authority (REQ-001-US-005-011)
        loop = asyncio.get_event_loop()
        if len(volks) > 1:
            # Check if multiple volks claim authority
            authorities = []
            for volk in volks:
                query = f"pam:pam_owns('{volk}', '{component}')"
                owns = await loop.run_in_executor(None, self._prolog.query_bool, query)
                if owns:
                    authorities.append(volk)

            if len(authorities) > 1:
                return Conflict(
                    conflict_id=f"conflict-{uuid.uuid4().hex[:8]}",
                    conflict_type=ConflictType.overlapping_authority,
                    component=component,
                    volks=authorities,
                    description=f"Multiple volks claim authority over {component}: {authorities}",
                )

        # Detect boundary violation (REQ-001-US-005-012)
        for volk in volks:
            query = f"pam:cross_boundary_allowed('{volk}', _, '{component}')"
            allowed = await loop.run_in_executor(None, self._prolog.query_bool, query)
            if not allowed:
                return Conflict(
                    conflict_id=f"conflict-{uuid.uuid4().hex[:8]}",
                    conflict_type=ConflictType.boundary_violation,
                    component=component,
                    volks=[volk],
                    description=f"Volk {volk} attempting boundary violation on {component}",
                )

        return None

    @traces(spec="001", story="US-005", reqs=["REQ-001-US-005-021"])
    async def create_resolution_workflow(self, conflict: Conflict) -> str:
        """Create conflict resolution workflow (REQ-001-US-005-013).

        Args:
            conflict: Conflict to resolve

        Returns:
            Workflow ID
        """
        workflow_id = f"workflow-{uuid.uuid4().hex[:8]}"

        workflow = {
            "workflow_id": workflow_id,
            "conflict_id": conflict.conflict_id,
            "status": ResolutionStatus.pending.value,
            "created_at": _utc_now().isoformat(),
            "assigned_to": None,
            "strategy": None,
        }

        self._workflows[workflow_id] = workflow

        try:
            # Store in Redis
            key = f"conflict:{workflow_id}"
            await self._redis.set(key, json.dumps(workflow))
        except Exception as e:
            logger.error(f"Failed to store workflow {workflow_id}: {e}")
            raise

        logger.info(
            f"Created resolution workflow {workflow_id} for conflict {conflict.conflict_id}"
        )

        return workflow_id

    async def assign_to_authority(self, workflow_id: str, authority: str) -> None:
        """Assign conflict to resolution authority (REQ-001-US-005-014).

        Args:
            workflow_id: Workflow ID
            authority: Authority to assign to
        """
        if workflow_id not in self._workflows:
            logger.warning(f"Workflow {workflow_id} not found")
            return

        self._workflows[workflow_id]["assigned_to"] = authority
        self._workflows[workflow_id]["status"] = ResolutionStatus.in_progress.value

        try:
            # Update Redis
            key = f"conflict:{workflow_id}"
            await self._redis.set(key, json.dumps(self._workflows[workflow_id]))
        except Exception as e:
            logger.error(f"Failed to assign workflow {workflow_id}: {e}")
            raise

        logger.info(f"Assigned workflow {workflow_id} to {authority}")

    async def track_resolution_status(
        self, workflow_id: str, status: ResolutionStatus
    ) -> None:
        """Track resolution workflow status (REQ-001-US-005-015).

        Args:
            workflow_id: Workflow ID
            status: New status
        """
        if workflow_id not in self._workflows:
            logger.warning(f"Workflow {workflow_id} not found")
            return

        self._workflows[workflow_id]["status"] = status.value
        self._workflows[workflow_id]["updated_at"] = _utc_now().isoformat()

        try:
            # Update Redis
            key = f"conflict:{workflow_id}"
            await self._redis.set(key, json.dumps(self._workflows[workflow_id]))
        except Exception as e:
            logger.error(f"Failed to track resolution status for {workflow_id}: {e}")
            raise

        logger.info(f"Workflow {workflow_id} status updated to {status.value}")

    @traces(spec="001", story="US-005", reqs=["REQ-001-US-005-021"])
    async def apply_resolution_strategy(
        self, conflict: Conflict, strategy: ResolutionStrategy
    ) -> Resolution:
        """Apply resolution strategy to conflict (REQ-001-US-005-016 to REQ-001-US-005-018).

        Args:
            conflict: Conflict to resolve
            strategy: Resolution strategy to apply

        Returns:
            Resolution result
        """
        resolution_id = f"res-{uuid.uuid4().hex[:8]}"
        workflow_id = await self.create_resolution_workflow(conflict)

        resolution_text = ""

        if strategy == ResolutionStrategy.vote:
            # Vote strategy (REQ-001-US-005-016)
            resolution_text = await self._apply_vote_strategy(conflict)
        elif strategy == ResolutionStrategy.priority:
            # Priority strategy (REQ-001-US-005-017)
            resolution_text = await self._apply_priority_strategy(conflict)
        elif strategy == ResolutionStrategy.delegation:
            # Delegation strategy (REQ-001-US-005-018)
            resolution_text = await self._apply_delegation_strategy(conflict)

        resolution = Resolution(
            resolution_id=resolution_id,
            workflow_id=workflow_id,
            resolved_by="system",
            resolution_text=resolution_text,
            strategy=strategy,
        )

        # Log resolution decision (REQ-001-US-005-019)
        await self._log_resolution_decision(conflict, resolution)

        # Archive to graph (REQ-001-US-005-020)
        if self._graph and self._graph._repo:
            # Create Resolution node in graph via a managed write transaction
            await self._graph._repo.execute_write(
                """
                CREATE (r:Resolution {
                    resolution_id: $resolution_id,
                    workflow_id: $workflow_id,
                    conflict_id: $conflict_id,
                    resolved_by: $resolved_by,
                    resolution_text: $resolution_text,
                    strategy: $strategy,
                    resolved_at: $resolved_at
                })
                """,
                {
                    "resolution_id": resolution.resolution_id,
                    "workflow_id": resolution.workflow_id,
                    "conflict_id": conflict.conflict_id,
                    "resolved_by": resolution.resolved_by,
                    "resolution_text": resolution.resolution_text,
                    "strategy": strategy.value,
                    "resolved_at": resolution.resolved_at.isoformat(),
                },
            )

            # Also log as audit event
            from orchestrata_ultimate.orchestrator.graph_audit import (  # noqa: PLC0415
                AuditEvent,
                AuditEventType,
            )

            event = AuditEvent(
                event_type=AuditEventType.ESCALATION,
                session_id=resolution_id,
                volk_id="conflict_resolver",
                timestamp=_utc_now(),
                data={
                    "conflict_id": conflict.conflict_id,
                    "resolution_id": resolution_id,
                    "strategy": strategy.value,
                    "resolution_text": resolution_text,
                    "action": "conflict_resolved",
                },
            )
            await self._graph.log_event(event)

        return resolution

    @traces(spec="001", story="US-005", reqs=["REQ-001-US-005-021"])
    async def _apply_vote_strategy(self, conflict: Conflict) -> str:
        """Apply vote resolution strategy (REQ-001-US-005-016).

        Args:
            conflict: Conflict to resolve

        Returns:
            Resolution description
        """
        # In real implementation, this would collect votes from stakeholders
        # For now, simulate majority vote
        volks = conflict.volks
        if volks:
            winner = volks[0]  # First volk wins in simulation
            return (
                f"Vote resolution: {winner} granted authority over {conflict.component}"
            )
        return "Vote resolution: No clear winner"

    @traces(spec="001", story="US-005", reqs=["REQ-001-US-005-021"])
    async def _apply_priority_strategy(self, conflict: Conflict) -> str:
        """Apply priority resolution strategy (REQ-001-US-005-017).

        Args:
            conflict: Conflict to resolve

        Returns:
            Resolution description
        """
        # Query Prolog for priority levels
        volks = conflict.volks
        if volks:
            # First volk has priority in simulation
            winner = volks[0]
            return f"Priority resolution: {winner} has higher priority for {conflict.component}"
        return "Priority resolution: No clear priority"

    @traces(spec="001", story="US-005", reqs=["REQ-001-US-005-021"])
    async def _apply_delegation_strategy(self, conflict: Conflict) -> str:
        """Apply delegation resolution strategy (REQ-001-US-005-018).

        Args:
            conflict: Conflict to resolve

        Returns:
            Resolution description
        """
        # Delegate to next level authority
        volks = conflict.volks
        if volks:
            delegated_to = "orchestrator"  # Default delegation target
            return (
                f"Delegation resolution: Delegated to {delegated_to} for final decision"
            )
        return "Delegation resolution: Unable to delegate"

    @traces(spec="001", story="US-005", reqs=["REQ-001-US-005-024"])
    async def _log_resolution_decision(
        self, conflict: Conflict, resolution: Resolution
    ) -> None:
        """Log resolution decision (REQ-001-US-005-019).

        Args:
            conflict: Conflict that was resolved
            resolution: Resolution applied
        """
        log_data = {
            "conflict_id": conflict.conflict_id,
            "resolution_id": resolution.resolution_id,
            "strategy": resolution.strategy.value,
            "resolution_text": resolution.resolution_text,
            "resolved_at": resolution.resolved_at.isoformat(),
        }

        try:
            key = f"resolution_log:{conflict.conflict_id}"
            await self._redis.set(key, json.dumps(log_data))
        except Exception as e:
            logger.error(
                f"Failed to log resolution decision for {conflict.conflict_id}: {e}"
            )
            raise

        logger.info(
            f"Resolution decision logged for conflict {conflict.conflict_id}: "
            f"{resolution.strategy.value}"
        )