orchestrata_ultimate.core.protocols.base

Reference › orchestrata_ultimate › core › protocols › base

The shared vocabulary of the component protocols, imported by components, memory and the dependency-injection container. Three enums classify components: ComponentType (kernel, postman, orchestrator, state manager, volk runtime, memory engine, provider manager, artifact manager, UI console, API gateway, generic service), HealthStatus (healthy, degraded, unhealthy, unknown, stopped) and ComponentScope (singleton, transient, request), which the DI container uses for lifecycles. Four dataclasses describe data: ComponentMetadata (id, name, version, type, description, dependencies, capabilities and timestamps), ServiceEndpoint, SystemEvent (with to_dict and from_dict) and EventSubscription (with expiry checking).

Two protocols follow. EventBusProtocol defines emit, subscribe with a topic pattern, unsubscribe, replay of past events and a subscriber_count. ComponentProtocol is the lifecycle contract every component implements: initialize, start, stop, get_health and get_metadata. base.service supplies default implementations.

At a glance

Class diagram

Dependency diagram

API

Classes

ComponentType (enum)

Bases: Enum · Line 24

Types of components in the system

HealthStatus (enum)

Bases: Enum · Line 40

Component health status

ComponentScope (enum)

Bases: Enum · Decorators: traces(spec='024', story='Core Registration', reqs=['REQ-024-002']) · Line 54

Component scope for dependency injection

ComponentMetadata (dataclass)

Decorators: dataclass · Line 68

Metadata for a component

ServiceEndpoint (dataclass)

Decorators: dataclass · Line 83

Service endpoint for discovery

SystemEvent (dataclass)

Decorators: dataclass · Line 94

System-wide event

EventSubscription (dataclass)

Decorators: dataclass · Line 136

Represents a subscription to events matching a topic pattern.

EventBusProtocol (protocol)

Bases: Protocol · Line 155

System-wide event publish/subscribe infrastructure.

ComponentProtocol (protocol)

Bases: Protocol · Decorators: runtime_checkable, traces(spec='011', story='Core Interface', reqs=['REQ-011-003']) · Line 203

Base protocol for all Orchestrata components.

Source

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

"""
Base component protocol types for Orchestrata Ultimate.

Foundational types and protocols shared across all core protocol modules.
This module has no intra-package or cross-package dependencies — it is the
canonical source for these base types; the legacy interfaces-facade
equivalent module re-exports them instead of redefining them.
"""

import uuid
from collections.abc import Awaitable, Callable
from dataclasses import dataclass, field
from datetime import UTC, datetime
from enum import Enum
from typing import Any, Protocol, runtime_checkable

from orchestrata_ultimate.core.base.traces import traces

# ============================================================================
# Base Types
# ============================================================================


class ComponentType(Enum):
    """Types of components in the system"""

    KERNEL = "kernel"
    POSTMAN = "postman"
    ORCHESTRATOR = "orchestrator"
    STATE_MANAGER = "state_manager"
    VOLK_RUNTIME = "volk_runtime"
    MEMORY_ENGINE = "memory_engine"
    PROVIDER_MANAGER = "provider_manager"
    ARTIFACT_MANAGER = "artifact_manager"
    UI_CONSOLE = "ui_console"
    API_GATEWAY = "api_gateway"
    SERVICE = "service"


class HealthStatus(Enum):
    """Component health status"""

    HEALTHY = "healthy"
    DEGRADED = "degraded"
    UNHEALTHY = "unhealthy"
    UNKNOWN = "unknown"
    STOPPED = "stopped"


REQUEST = "request"


@traces(spec="024", story="Core Registration", reqs=["REQ-024-002"])
class ComponentScope(Enum):
    """Component scope for dependency injection"""

    SINGLETON = "singleton"
    TRANSIENT = "transient"
    REQUEST = "request"


# ============================================================================
# Data Classes
# ============================================================================


@dataclass
class ComponentMetadata:
    """Metadata for a component"""

    id: str
    name: str
    version: str
    component_type: ComponentType
    description: str = ""
    dependencies: list[str] = field(default_factory=list)
    capabilities: list[str] = field(default_factory=list)
    created_at: datetime = field(default_factory=lambda: datetime.now(UTC))
    updated_at: datetime = field(default_factory=lambda: datetime.now(UTC))


@dataclass
class ServiceEndpoint:
    """Service endpoint for discovery"""

    service_type: str
    endpoint: str
    protocol: str  # "grpc", "http", "ws", "amqp"
    health_check: str = "/health"
    metadata: dict[str, Any] = field(default_factory=dict)


@dataclass
class SystemEvent:
    """System-wide event"""

    id: str = field(default_factory=lambda: str(uuid.uuid4()))
    type: str = ""
    source: str = ""
    timestamp: datetime = field(default_factory=lambda: datetime.now(UTC))
    payload: dict[str, Any] = field(default_factory=dict)
    correlation_id: str | None = None
    severity: str = "info"

    def to_dict(self) -> dict[str, Any]:
        """Convert to dictionary for serialization"""
        return {
            "id": self.id,
            "type": self.type,
            "source": self.source,
            "timestamp": self.timestamp.isoformat(),
            "payload": self.payload,
            "correlation_id": self.correlation_id,
            "severity": self.severity,
        }

    @classmethod
    def from_dict(cls, data: dict[str, Any]) -> SystemEvent:
        """Create from dictionary"""
        try:
            timestamp = datetime.fromisoformat(data["timestamp"])
        except ValueError, KeyError:
            timestamp = datetime.now(UTC)
        return cls(
            id=data.get("id", str(uuid.uuid4())),
            type=data["type"],
            source=data["source"],
            timestamp=timestamp,
            payload=data.get("payload", {}),
            correlation_id=data.get("correlation_id"),
            severity=data.get("severity", "info"),
        )


@dataclass
class EventSubscription:
    """Represents a subscription to events matching a topic pattern."""

    id: str
    topic_pattern: str
    subscriber_id: str
    created_at: datetime
    expires_at: datetime | None = None
    event_count: int = 0

    def is_expired(self, reference: datetime | None = None) -> bool:
        """Check if the subscription has expired."""
        if self.expires_at is None:
            return False
        if reference is None:
            reference = datetime.now(UTC)
        return reference >= self.expires_at


class EventBusProtocol(Protocol):
    """System-wide event publish/subscribe infrastructure."""

    async def emit(self, event: SystemEvent) -> None:
        """Emit an event to all matching subscribers."""
        ...

    @traces(spec="015", story="US-011", reqs=["REQ-015-149"])
    async def subscribe(
        self,
        topic_pattern: str,
        handler: Callable[[SystemEvent], Awaitable[None]],
        *,
        subscriber_id: str | None = None,
        ttl_seconds: int | None = None,
        filter_fn: Callable[[SystemEvent], bool] | None = None,
    ) -> EventSubscription:
        """Subscribe to events matching a topic pattern."""
        ...

    async def unsubscribe(self, subscription_id: str) -> bool:
        """Unsubscribe by subscription ID."""
        ...

    async def replay(
        self,
        topic_pattern: str,
        handler: Callable[[SystemEvent], Awaitable[None]],
        *,
        since: datetime | None = None,
        until: datetime | None = None,
        count: int | None = None,
    ) -> int:
        """Replay historical events matching a topic pattern."""
        ...

    def subscriber_count(self, topic_pattern: str | None = None) -> int:
        """Count subscribers matching the pattern, or all if None."""
        ...


# ============================================================================
# Base Protocol
# ============================================================================


@runtime_checkable
@traces(spec="011", story="Core Interface", reqs=["REQ-011-003"])
class ComponentProtocol(Protocol):
    """
    Base protocol for all Orchestrata components.

    All components must implement this protocol to ensure
    consistent lifecycle management and interoperability.
    """

    async def initialize(self, config: dict[str, Any]) -> None:
        """
        Initialize component with configuration.

        Args:
            config: Component configuration dictionary
        """
        ...

    async def start(self) -> None:
        """Start component operation."""
        ...

    async def stop(self) -> None:
        """Stop component gracefully."""
        ...

    async def get_health(self) -> HealthStatus:
        """Get component health status."""
        ...

    def get_metadata(self) -> ComponentMetadata:
        """Get component metadata."""
        ...


__all__ = [
    "ComponentType",
    "HealthStatus",
    "REQUEST",
    "ComponentScope",
    "ComponentMetadata",
    "ServiceEndpoint",
    "SystemEvent",
    "EventSubscription",
    "EventBusProtocol",
    "ComponentProtocol",
]