orchestrata_ultimate.agents.coordinator

Reference › orchestrata_ultimate › agents › coordinator

coordinator composes many task executions into multi-volk coordination patterns. VolkCoordinator is constructed with a TaskExecutor (task_executor) and a MultiagentPrologBridge (prolog_bridge), and implements the IVolkCoordinator contract from interfaces. It raises the coordination errors from exceptions. AgentCoordinator is an alias subclass.

• elect_leader asserts election facts into Prolog (_assert_election_facts) and returns the first candidate the rules find eligible (_find_first_eligible_leader), or None.
• propose_consensus asserts the voter group and each vote (_assert_consensus_group, _cast_votes) and then asks Prolog for quorum and majority in _tally_consensus, which builds a ConsensusResult.
• fan_out submits copies of one task to several aliases concurrently (_submit_task_for_alias) and gathers the results; _convert_fan_out_results turns exceptions into failed TaskResult objects instead of aborting the batch.
• pipeline runs a list of stages sequentially through the executor and returns the final TaskResult.

The decision logic therefore lives in the Prolog rules; this module only asserts facts and reads answers.

At a glance

Class diagram

Dependency diagram

API

Classes

VolkCoordinator

Line 31

Multi-Volk coordination patterns.

AgentCoordinator

Bases: VolkCoordinator · Line 291

Coordinator alias for VolkCoordinator.

Source

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

# mypy: ignore-errors
"""Volk Coordinator — multi-Volk coordination patterns.

Implements leader election, consensus, fan-out, and pipeline
patterns using Prolog governance and TaskExecutor.

Author: Mohamed
Spec: 005-multi-agent (amended)
"""

import asyncio
import logging
import time

from orchestrata_ultimate.agents.exceptions import (
    ConsensusError,
    CoordinationError,
    LeaderElectionError,
)
from orchestrata_ultimate.agents.prolog_bridge import MultiagentPrologBridge
from orchestrata_ultimate.agents.task_executor import TaskExecutor
from orchestrata_ultimate.agents.types import (
    ConsensusResult,
    TaskDefinition,
    TaskResult,
)

_logger = logging.getLogger(__name__)


class VolkCoordinator:
    """Multi-Volk coordination patterns.

    Provides leader election, consensus, fan-out, and pipeline
    patterns backed by Prolog governance rules.

    Parameters
    ----------
    executor : TaskExecutor
        Task executor for dispatching work.
    prolog : MultiagentPrologBridge
        Bridge for coordination governance queries.
    """

    def __init__(
        self,
        executor: TaskExecutor,
        prolog: MultiagentPrologBridge,
    ) -> None:
        self._executor = executor
        self._prolog = prolog

    # ================================================================
    # Leader Election
    # ================================================================

    def _assert_election_facts(self, group_id: str, candidates: list[str]) -> None:
        """Assert Prolog facts for a leader election group."""
        self._prolog.assert_group_size(group_id, len(candidates))
        for cid in candidates:
            self._prolog.assert_volk_in_group(cid, group_id)
            self._prolog.assert_volk_running(cid)
            self._prolog.assert_volk_eligible_for_leadership(cid)
            self._prolog.assert_volk_priority(cid, "high")

    def _find_first_eligible_leader(
        self, group_id: str, candidates: list[str], start_time: float
    ) -> str | None:
        """Return the first candidate eligible for leadership, or None."""
        for cid in candidates:
            if self._prolog.can_become_leader(cid, group_id):
                elapsed_ms = (time.monotonic() - start_time) * 1000
                _logger.info(
                    f"Leader elected: {cid} for group {group_id} in {elapsed_ms:.1f}ms"
                )
                return cid
        _logger.info(f"No leader elected for group {group_id}")
        return None

    async def elect_leader(
        self,
        group_id: str,
        candidates: list[str],
    ) -> str | None:
        """Elect a leader from candidate instance IDs.

        Sets up Prolog group facts, checks each candidate's eligibility
        via can_become_leader/2, and returns the first eligible.

        Returns None if no candidate is eligible.

        Raises
        ------
        LeaderElectionError
            If the election process encounters an error.
        """
        if not candidates:
            raise LeaderElectionError(
                "No candidates provided for leader election",
                group_id=group_id,
            )

        start_time = time.monotonic()

        try:
            self._assert_election_facts(group_id, candidates)
            return self._find_first_eligible_leader(group_id, candidates, start_time)
        except LeaderElectionError:
            raise
        except Exception as e:
            raise LeaderElectionError(
                f"Leader election failed for group '{group_id}': {e}",
                group_id=group_id,
            )

    # ================================================================
    # Consensus
    # ================================================================

    def _assert_consensus_group(self, group_id: str, voters: list[str]) -> None:
        """Assert Prolog group facts for consensus voters."""
        self._prolog.assert_group_size(group_id, len(voters))
        for vid in voters:
            self._prolog.assert_volk_in_group(vid, group_id)

    def _cast_votes(self, group_id: str, actual_votes: dict[str, str]) -> None:
        """Assert each voter's vote in Prolog."""
        for vid, vote in actual_votes.items():
            self._prolog.assert_volk_votes(vid, group_id, vote)

    def _tally_consensus(
        self,
        group_id: str,
        actual_votes: dict[str, str],
        total_voters: int,
    ) -> ConsensusResult:
        """Query Prolog for quorum/majority and build a ConsensusResult."""
        quorum = self._prolog.quorum_met(group_id, len(actual_votes))
        yes_count = sum(1 for v in actual_votes.values() if v == "yes")
        no_count = sum(1 for v in actual_votes.values() if v == "no")
        majority = self._prolog.majority_reached(group_id, "yes")
        decision = "yes" if majority else "no"
        return ConsensusResult(
            group_id=group_id,
            decision=decision,
            votes_for=yes_count,
            votes_against=no_count,
            total_voters=total_voters,
            quorum_met=quorum,
        )

    async def propose_consensus(
        self,
        group_id: str,
        proposal: str,
        voters: list[str],
        votes: dict[str, str] | None = None,
    ) -> ConsensusResult:
        """Run a consensus vote.

        If votes are provided, uses them directly. Otherwise,
        assumes all voters vote 'yes' (for automated consensus).

        Raises
        ------
        ConsensusError
            If the consensus process fails.
        """
        if not voters:
            raise ConsensusError(
                "No voters provided for consensus",
                group_id=group_id,
            )

        try:
            self._assert_consensus_group(group_id, voters)
            actual_votes = votes or {v: "yes" for v in voters}
            self._cast_votes(group_id, actual_votes)
            return self._tally_consensus(group_id, actual_votes, len(voters))
        except ConsensusError:
            raise
        except Exception as e:
            raise ConsensusError(
                f"Consensus failed for group '{group_id}': {e}",
                group_id=group_id,
            )

    # ================================================================
    # Fan-Out
    # ================================================================

    async def _submit_task_for_alias(
        self, task: TaskDefinition, alias: str
    ) -> TaskResult:
        """Copy a task definition for a specific alias and submit it."""
        task_copy = TaskDefinition(
            action=task.action,
            target=alias,
            parameters=task.parameters.copy(),
            required_capabilities=task.required_capabilities.copy(),
            priority=task.priority,
            timeout_seconds=task.timeout_seconds,
            requester=task.requester,
            correlation_id=task.correlation_id,
            metadata=task.metadata.copy(),
        )
        return await self._executor.submit(task_copy)

    def _convert_fan_out_results(
        self, raw_results: list, aliases: list[str]
    ) -> list[TaskResult]:
        """Convert gather results, turning exceptions into failed TaskResults."""
        processed: list[TaskResult] = []
        for i, result in enumerate(raw_results):
            if isinstance(result, Exception):
                processed.append(
                    TaskResult(
                        task_id=f"fan-out-{aliases[i]}",
                        success=False,
                        error_message=str(result),
                    )
                )
            else:
                processed.append(result)
        return processed

    async def fan_out(
        self,
        task: TaskDefinition,
        aliases: list[str],
    ) -> list[TaskResult]:
        """Execute a task across multiple Volks concurrently.

        Creates a copy of the task for each alias and submits all
        concurrently via asyncio.gather().
        """
        if not aliases:
            return []

        results = await asyncio.gather(
            *[self._submit_task_for_alias(task, alias) for alias in aliases],
            return_exceptions=True,
        )

        return self._convert_fan_out_results(list(results), aliases)

    # ================================================================
    # Pipeline
    # ================================================================

    async def pipeline(
        self,
        stages: list[tuple],
    ) -> TaskResult:
        """Execute tasks sequentially through a pipeline.

        Each stage is (alias, task_definition). Output of stage N
        becomes input parameters of stage N+1.

        Returns the final stage's result.
        """
        if not stages:
            raise CoordinationError("No stages provided for pipeline")

        last_result: TaskResult | None = None

        for i, (alias, stage_task) in enumerate(stages):
            # Inject previous stage output as parameters
            if last_result is not None and last_result.output:
                if isinstance(last_result.output, dict):
                    stage_task.parameters["previous_output"] = last_result.output

            stage_task.target = alias
            stage_task.metadata["pipeline_stage"] = i

            last_result = await self._executor.submit(stage_task)

            if not last_result.success:
                _logger.warning(
                    f"Pipeline stage {i} ({alias}) failed: {last_result.error_message}"
                )
                return last_result

        # stages is non-empty (checked above), so the loop above always ran
        # at least once and last_result was assigned a real TaskResult.
        assert last_result is not None
        return last_result


# Class alias for orchestrata-ultimate naming
class AgentCoordinator(VolkCoordinator):
    """Coordinator alias for VolkCoordinator."""

    pass