orchestrata_ultimate.agents.interfaces

Reference › orchestrata_ultimate › agents › interfaces

interfaces declares the Multi-Agent Runtime's contracts as runtime-checkable protocols, so components can be swapped or mocked without inheritance. It imports the data types from types and the core protocols, and the package __init__ re-exports the four protocols.

IVolkRuntime is the instance lifecycle contract: spawn, terminate, force_terminate, get_instance, list_instances and get_instance_health. ITaskExecutor covers submit, cancel, get_task and list_tasks. IVolkPool covers acquire, release, get_pool_status and set_pool_config. IVolkCoordinator covers elect_leader, propose_consensus and fan_out. All methods are asynchronous. The concrete implementations are runtime, task_executor, pool and coordinator. Only IVolkRuntime carries a traces marker (specification 005, requirement FR3).

At a glance

Class diagram

Dependency diagram

API

Classes

IVolkRuntime (protocol)

Bases: Protocol · Decorators: runtime_checkable, traces(spec='005', story='US-003', reqs=['REQ-005-FR3-001']) · Line 29

Manages Volk instance lifecycle.

ITaskExecutor (protocol)

Bases: Protocol · Decorators: runtime_checkable · Line 87

Submits and tracks task execution on Volk instances.

IVolkPool (protocol)

Bases: Protocol · Decorators: runtime_checkable · Line 126

Manages per-alias Volk instance pools.

IVolkCoordinator (protocol)

Bases: Protocol · Decorators: runtime_checkable · Line 159

Multi-Volk coordination patterns.

Source

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

"""Multi-Agent Runtime interfaces.

Runtime-checkable protocols for Volk lifecycle, task execution,
pool management, and multi-Volk coordination.

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

# Spec: 005 - Multi-Agent System

from typing import Any, Protocol, runtime_checkable

from orchestrata_ultimate.agents.types import (
    ConsensusResult,
    InstanceStatus,
    PoolConfig,
    PoolStatus,
    TaskDefinition,
    TaskResult,
    VolkInstance,
)
from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.core.protocols import HealthStatus


@runtime_checkable
@traces(spec="005", story="US-003", reqs=["REQ-005-FR3-001"])
class IVolkRuntime(Protocol):
    """Manages Volk instance lifecycle.

    Responsible for spawning, terminating, and tracking running
    Volk instances. Implements IComponent with ComponentType.VOLK_RUNTIME.
    """

    @traces(spec="005", story="US-001", reqs=["REQ-005-US-001-001"])
    async def spawn(self, alias: str, **kwargs: Any) -> VolkInstance:
        """Spawn a new Volk instance for the given alias.

        Resolves the alias via VolkRegistry, checks Prolog governance
        rules, asserts runtime facts, and persists instance state.

        Raises:
            VolkSpawnError: If governance denies or spawn fails.
        """
        ...  # pragma: no cover

    async def terminate(self, instance_id: str, force: bool = False) -> bool:
        """Terminate a running Volk instance.

        Graceful termination drains current tasks first.
        Force termination stops immediately.

        Returns True if termination succeeded.

        Raises:
            VolkTerminateError: If governance denies termination.
        """
        ...  # pragma: no cover

    async def force_terminate(self, instance_id: str, reason: str) -> bool:
        """Force-terminate an instance with an emergency reason.

        Bypasses graceful shutdown. Used for resource exhaustion,
        security violations, and unresponsive instances.
        """
        ...  # pragma: no cover

    async def get_instance(self, instance_id: str) -> VolkInstance | None:
        """Get instance by ID. Returns None if not found."""
        ...  # pragma: no cover

    async def list_instances(
        self,
        alias: str | None = None,
        status: InstanceStatus | None = None,
    ) -> list[VolkInstance]:
        """List instances, optionally filtered by alias and/or status."""
        ...  # pragma: no cover

    async def get_instance_health(self, instance_id: str) -> HealthStatus:
        """Get health status of a specific instance."""
        ...  # pragma: no cover


@runtime_checkable
class ITaskExecutor(Protocol):
    """Submits and tracks task execution on Volk instances."""

    async def submit(self, task: TaskDefinition) -> TaskResult:
        """Submit a task for execution.

        Selects the best Volk via Prolog, acquires a pool instance,
        executes the task, and returns the result.

        Raises:
            TaskAssignmentError: If no suitable Volk is found.
            TaskExecutionError: If execution fails.
        """
        ...  # pragma: no cover

    async def cancel(self, task_id: str) -> bool:
        """Cancel a pending or executing task.

        Returns True if cancellation succeeded.

        Raises:
            TaskCancellationError: If the task cannot be cancelled.
        """
        ...  # pragma: no cover

    async def get_task(self, task_id: str) -> TaskDefinition | None:
        """Get task definition by ID. Returns None if not found."""
        ...  # pragma: no cover

    async def list_tasks(
        self,
        status: str | None = None,
        alias: str | None = None,
    ) -> list[TaskDefinition]:
        """List tasks, optionally filtered by status and/or executor alias."""
        ...  # pragma: no cover


@runtime_checkable
class IVolkPool(Protocol):
    """Manages per-alias Volk instance pools."""

    async def acquire(
        self,
        alias: str,
        task: TaskDefinition | None = None,
    ) -> VolkInstance:
        """Acquire an instance from the pool for the given alias.

        If no ready instance is available and capacity allows, spawns
        a new one. Uses Prolog least_loaded_volk/2 for selection.

        Raises:
            PoolExhaustedError: If pool is at capacity and all busy.
        """
        ...  # pragma: no cover

    async def release(self, instance_id: str) -> None:
        """Release an instance back to the pool."""
        ...  # pragma: no cover

    @traces(spec="005", story="US-004", reqs=["REQ-005-AC6-001"])
    async def get_pool_status(self, alias: str) -> PoolStatus:
        """Get current pool status for the given alias."""
        ...  # pragma: no cover

    async def set_pool_config(self, config: PoolConfig) -> None:
        """Set or update pool configuration for an alias."""
        ...  # pragma: no cover


@runtime_checkable
class IVolkCoordinator(Protocol):
    """Multi-Volk coordination patterns."""

    async def elect_leader(
        self,
        group_id: str,
        candidates: list[str],
    ) -> str | None:
        """Elect a leader from candidates for the given group.

        Uses Prolog can_become_leader/2 for eligibility checks.
        Returns the elected leader's instance_id, or None if election fails.

        Raises:
            LeaderElectionError: If election fails.
        """
        ...  # pragma: no cover

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

        Asserts volk_votes facts, checks quorum_met/2 and
        majority_reached/2 via Prolog.

        Raises:
            ConsensusError: If consensus process fails.
        """
        ...  # pragma: no cover

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

        Uses asyncio.gather() with TaskExecutor.submit() for each alias.
        Returns results from all Volks.
        """
        ...  # pragma: no cover

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

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

        Returns the final stage's result.
        """
        ...  # pragma: no cover