orchestrata_ultimate.core.protocols.components

Reference › orchestrata_ultimate › core › protocols › components

The four top-level component contracts, each built on ComponentProtocol from base. Other packages refer to the subsystems through these protocols and not through their concrete classes; interfaces.components keeps the old I names for them.

• OrchestrationKernelProtocol manages components and services: register_component, unregister_component, get_component, list_components (optionally by type), discover_service, register_service and emit_event.
• PostmanProtocol is the message router: route a CXP packet, enqueue with a priority, dequeue for a volk, get_queue_status, acknowledge and retry_failed.
• OrchestratorProtocol manages volks and policy: hire_volk, fire_volk, list_volks, evaluate_policy and escalate.
• StateManagerProtocol is the keyed state store: get_state, set_state, update_state, delete_state, create_namespace and list_namespaces.

All methods are coroutines, and the module uses forward-reference sentinels for types such as the CXP packet to avoid circular imports. The persistent implementation of the state contract is state.state_manager.

At a glance

Class diagram

Dependency diagram

API

Classes

OrchestrationKernelProtocol (protocol)

Bases: ComponentProtocol, Protocol · Line 38

Central orchestration engine.

PostmanProtocol (protocol)

Bases: ComponentProtocol, Protocol · Decorators: traces(spec='004', story='US-INTG', reqs=['REQ-004-INTG-010', 'REQ-004-INTG-012']) · Line 129

Message routing and delivery engine (U:PS).

OrchestratorProtocol (protocol)

Bases: ComponentProtocol, Protocol · Line 222

Governance and policy management engine (U:Ox).

StateManagerProtocol (protocol)

Bases: ComponentProtocol, Protocol · Line 294

Persistent state management.

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/core/protocols/components.py

"""
Core component protocols for Orchestrata Ultimate.

Defines the kernel, postman, orchestrator, and state-manager protocols.

Depends on :mod:`orchestrata_ultimate.core.protocols.base` for the shared
base types and on :mod:`orchestrata_ultimate.core.packet` for the packet
vocabulary (Layer 0). This module imports nothing from the concrete
``cxp``/``postman``/``orchestrator``/``kernel`` packages — not even under
``TYPE_CHECKING``. Payload types owned by higher layers (postman's
``DeliveryResult``/``QueueEntry``/``QueueStatus``, orchestrator's
``VolkDefinition``/``VolkInstance``/``PolicyContext``/``PolicyDecision``/
``EscalationIssue``/``EscalationResult``) cannot be named here without an
upward static edge, so they appear as ``Any`` — the same convention as
``core.protocols.session`` and ``core.protocols.decoupling``; structural
conformance is provided by the concrete implementations
(``postman.postman.Postman``, ``orchestrator.orchestrator``).
"""

from __future__ import annotations

from typing import Any, Protocol

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.core.packet import CXPPacket
from orchestrata_ultimate.core.protocols.base import (
    ComponentProtocol,
    ComponentType,
    ServiceEndpoint,
    SystemEvent,
)

# ============================================================================
# Core Component Protocols
# ============================================================================


class OrchestrationKernelProtocol(ComponentProtocol, Protocol):
    """
    Central orchestration engine.

    Manages all system components, handles lifecycle events,
    and provides essential services like service discovery.
    """

    async def register_component(self, component: ComponentProtocol) -> str:
        """
        Register a component with the kernel.

        Args:
            component: Component to register

        Returns:
            Component ID assigned by kernel
        """
        ...

    async def unregister_component(self, component_id: str) -> None:
        """
        Unregister component from kernel.

        Args:
            component_id: ID of component to unregister
        """
        ...

    async def get_component(self, component_id: str) -> ComponentProtocol | None:
        """
        Get component by ID.

        Args:
            component_id: Component ID

        Returns:
            Component instance or None if not found
        """
        ...

    async def list_components(
        self, filter_type: ComponentType | None = None
    ) -> list[ComponentProtocol]:
        """
        List all registered components.

        Args:
            filter_type: Optional component type filter

        Returns:
            List of component instances
        """
        ...

    async def discover_service(self, service_type: str) -> list[ServiceEndpoint]:
        """
        Discover services of given type.

        Args:
            service_type: Type of service to discover

        Returns:
            List of service endpoints
        """
        ...

    async def register_service(self, endpoint: ServiceEndpoint) -> None:
        """
        Register service endpoint for discovery.

        Args:
            endpoint: Service endpoint to register
        """
        ...

    async def emit_event(self, event: SystemEvent) -> None:
        """
        Emit system-wide event.

        Args:
            event: Event to emit
        """
        ...


@traces(
    spec="004",
    story="US-INTG",
    reqs=["REQ-004-INTG-010", "REQ-004-INTG-012"],
)
class PostmanProtocol(ComponentProtocol, Protocol):
    """
    Message routing and delivery engine (U:PS).

    Responsible for delivering CXP packets between Volks
    with guaranteed delivery semantics.
    """

    @traces(spec="002", story="US1", reqs=["REQ-002-US-001-002"])
    async def route(self, packet: CXPPacket) -> Any:
        """
        Route CXP packet to destination.

        Args:
            packet: CXP packet to route

        Returns:
            Delivery result (postman's ``DeliveryResult``)
        """
        ...

    @traces(spec="002", story="FR4", reqs=["REQ-002-FR4-006"])
    async def enqueue(self, packet: CXPPacket, priority: int = 0) -> str:
        """
        Enqueue packet for delivery.

        Args:
            packet: CXP packet to enqueue
            priority: Delivery priority (higher = more important)

        Returns:
            Queue entry ID
        """
        ...

    @traces(spec="002", story="FR4", reqs=["REQ-002-FR4-007"])
    async def dequeue(self, volk_id: str) -> Any | None:
        """
        Dequeue next packet for Volk.

        NOTE: the concrete ``Postman.dequeue`` (postman.postman.Postman)
        returns the lightweight ``QueueEntry`` (packet_id/priority), not the
        full ``CXPPacket`` — this annotation was stale/aspirational relative
        to the real, tested implementation (tests/postman/test_postman.py::
        test_enqueue_dequeue asserts ``entry.packet_id``). The return type
        appears as ``Any`` because ``QueueEntry`` lives in Layer 2 and cannot
        be named in Layer 0 without an upward static edge.

        Args:
            volk_id: Volk ID to dequeue for

        Returns:
            Next queue entry, or None if queue empty
        """
        ...

    @traces(spec="002", story="US3", reqs=["REQ-002-US-003-004"])
    async def get_queue_status(self, volk_id: str | None = None) -> Any:
        """
        Get queue status for Volk(s).

        Args:
            volk_id: Optional specific Volk ID

        Returns:
            Queue status information (postman's ``QueueStatus``)
        """
        ...

    @traces(spec="002", story="US3", reqs=["REQ-002-US-003-015"])
    async def acknowledge(self, packet_id: str) -> None:
        """
        Acknowledge packet delivery.

        Args:
            packet_id: Packet ID to acknowledge
        """
        ...

    @traces(spec="002", story="US1", reqs=["REQ-002-US-001-024"])
    async def retry_failed(self, packet_id: str) -> Any:
        """
        Retry failed delivery.

        Args:
            packet_id: Packet ID to retry

        Returns:
            New delivery result (postman's ``DeliveryResult``)
        """
        ...


class OrchestratorProtocol(ComponentProtocol, Protocol):
    """
    Governance and policy management engine (U:Ox).

    Manages staffing, governance, escalation policies,
    and routing decisions for the entire system.
    """

    @traces(spec="001", story="US-001", reqs=["REQ-001-001", "REQ-001-008"])
    async def hire_volk(self, definition: Any) -> Any:
        """
        Hire new Volk with given definition.

        Args:
            definition: Volk definition (orchestrator's ``VolkDefinition``)

        Returns:
            Volk instance (orchestrator's ``VolkInstance``)
        """
        ...

    @traces(spec="001", story="US-001", reqs=["REQ-001-021"])
    async def fire_volk(self, volk_id: str, reason: str | None = None) -> None:
        """
        Terminate Volk operation.

        Args:
            volk_id: Volk ID to terminate
            reason: Optional termination reason
        """
        ...

    @traces(spec="001", story="US-001", reqs=["REQ-001-011"])
    async def list_volks(self, filter_status: str | None = None) -> list[Any]:
        """
        List all Volks.

        Args:
            filter_status: Optional status filter

        Returns:
            List of Volk instances (orchestrator's ``VolkInstance``)
        """
        ...

    @traces(spec="001", story="US-003", reqs=["REQ-001-US-003-002"])
    async def evaluate_policy(self, context: Any) -> Any:
        """
        Evaluate governance policy.

        Args:
            context: Policy evaluation context (orchestrator's ``PolicyContext``)

        Returns:
            Policy decision (orchestrator's ``PolicyDecision``)
        """
        ...

    @traces(spec="001", story="US-005", reqs=["REQ-001-US-005-012"])
    async def escalate(self, issue: Any) -> Any:
        """
        Escalate issue through governance chain.

        Args:
            issue: Issue to escalate (orchestrator's ``EscalationIssue``)

        Returns:
            Escalation result (orchestrator's ``EscalationResult``)
        """
        ...


class StateManagerProtocol(ComponentProtocol, Protocol):
    """
    Persistent state management.

    Manages state for Volks, threads, and system configuration
    with support for multiple storage backends.
    """

    async def get_state(self, key: str, default: Any = None) -> Any:
        """
        Get state value.

        Args:
            key: State key
            default: Default value if key not found

        Returns:
            State value
        """
        ...

    async def set_state(self, key: str, value: Any) -> None:
        """
        Set state value.

        Args:
            key: State key
            value: Value to set
        """
        ...

    async def update_state(self, key: str, updater: Any) -> Any:
        """
        Update state value using updater function.

        Args:
            key: State key
            updater: Function that takes current value and returns new value

        Returns:
            Updated value
        """
        ...

    async def delete_state(self, key: str) -> None:
        """
        Delete state value.

        Args:
            key: State key to delete
        """
        ...

    async def create_namespace(self, name: str) -> None:
        """
        Create state namespace.

        Args:
            name: Namespace name
        """
        ...

    async def list_namespaces(self) -> list[str]:
        """
        List all namespaces.

        Returns:
            List of namespace names
        """
        ...


__all__ = [
    "OrchestrationKernelProtocol",
    "PostmanProtocol",
    "OrchestratorProtocol",
    "StateManagerProtocol",
]