orchestrata_ultimate.cxp.interfaces

Reference › orchestrata_ultimate › cxp › interfaces

interfaces declares the structural contracts of the CXP package as typing.Protocol classes, so components depend on behaviour and not on concrete classes. IPacket lists the read-only view of a packet: message id, sender and recipient, thread id, subject, payload, timestamp, priority, delivery and work status, protocol version, get_headers and validate. ISerializer covers to_json, from_json, to_bytes and from_bytes; the concrete serializers are json_serializer, binary_serializer and serializer. IValidator covers packet and schema validation and registration of field validators (validator). IThreadManager covers create, add, read, close and reopen of threads (threads). IDeliveryTracker covers status setting, history, failure recording and retry counting (delivery_status). IPrologBridge covers asserting and retracting packets and querying them by thread or status (prolog_bridge). The package __init__ re-exports them. See cxp.

At a glance

Class diagram

Dependency diagram

API

Classes

IPacket (protocol)

Bases: Protocol · Decorators: runtime_checkable, traces(spec='004', story='US-001', reqs=['REQ-004-US-001-025']) · Line 45

Interface for CXP packets (Spec 004 Section 2).

ISerializer (protocol)

Bases: Protocol · Decorators: runtime_checkable · Line 158

Interface for CXP packet serialization (Spec 004 Section 6).

IValidator (protocol)

Bases: Protocol · Decorators: runtime_checkable, traces(spec='004', story='US-007', reqs=['REQ-004-US-007-018']) · Line 203

Interface for CXP packet validation (Spec 004 Section 7).

IThreadManager (protocol)

Bases: Protocol · Decorators: runtime_checkable · Line 231

Interface for CXP thread management (Spec 004 Section 8).

IDeliveryTracker (protocol)

Bases: Protocol · Decorators: runtime_checkable · Line 270

Interface for CXP delivery tracking (Spec 004 Section 4).

IPrologBridge (protocol)

Bases: Protocol · Decorators: runtime_checkable · Line 298

Interface for CXP Prolog integration (Spec 004 Section 10).

Source

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

"""
CXP Protocol Interfaces.

Protocol (structural typing) classes defining the CXP contract.
These use typing.Protocol for static type checking without inheritance.

Spec 004 defines 6 interfaces:
  IPacket, ISerializer, IValidator,
  IThreadManager, IDeliveryTracker, IPrologBridge
"""

# Spec: 004 - CXP Protocol

from collections.abc import Callable
from datetime import datetime
from typing import (
    Any,
    Protocol,
    runtime_checkable,
)

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.core.packet import (
    DeliveryStatus,
    ValidationResult,
    WorkStatus,
)

__all__ = [
    "IPacket",
    "ISerializer",
    "IValidator",
    "IThreadManager",
    "IDeliveryTracker",
    "IPrologBridge",
]


@runtime_checkable
@traces(
    spec="004",
    story="US-001",
    reqs=["REQ-004-US-001-025"],
)
class IPacket(Protocol):
    """Interface for CXP packets (Spec 004 Section 2).

    Defines the 11 core fields every packet implementation must expose.
    Performance target: header extraction < 1ms.
    """

    @property
    @traces(
        spec="004",
        story="US-001",
        reqs=["REQ-004-US-001-002"],
    )
    def message_id(self) -> str: ...  # pragma: no cover

    @property
    @traces(
        spec="004",
        story="US-001",
        reqs=["REQ-004-US-001-003"],
    )
    def from_agent(self) -> str: ...  # pragma: no cover

    @property
    @traces(
        spec="004",
        story="US-001",
        reqs=["REQ-004-US-001-004"],
    )
    def to_agent(self) -> str: ...  # pragma: no cover

    @property
    @traces(
        spec="004",
        story="US-001",
        reqs=["REQ-004-US-001-005"],
    )
    def thread_id(self) -> str: ...  # pragma: no cover

    @property
    @traces(
        spec="004",
        story="US-001",
        reqs=["REQ-004-US-001-006"],
    )
    def subject(self) -> str: ...  # pragma: no cover

    @property
    @traces(
        spec="004",
        story="US-001",
        reqs=["REQ-004-US-001-007"],
    )
    def payload(self) -> dict[str, Any]: ...  # pragma: no cover

    @property
    @traces(
        spec="004",
        story="US-001",
        reqs=["REQ-004-US-001-008"],
    )
    def timestamp(self) -> datetime: ...  # pragma: no cover

    @property
    @traces(
        spec="004",
        story="US-001",
        reqs=["REQ-004-US-001-009"],
    )
    def priority(self) -> int: ...  # pragma: no cover

    @property
    @traces(
        spec="004",
        story="US-001",
        reqs=["REQ-004-US-001-010"],
    )
    def delivery_status(self) -> DeliveryStatus: ...  # pragma: no cover

    @property
    @traces(
        spec="004",
        story="US-001",
        reqs=["REQ-004-US-001-011"],
    )
    def work_status(self) -> WorkStatus: ...  # pragma: no cover

    @property
    @traces(
        spec="004",
        story="US-001",
        reqs=["REQ-004-US-001-012"],
    )
    def protocol_version(self) -> str: ...  # pragma: no cover

    @traces(spec="004", story="US-002", reqs=["REQ-004-US-002-001"])
    def get_headers(self) -> dict[str, Any]:
        """Extract all header fields as a dict.

        Performance target: < 1ms.
        """
        ...  # pragma: no cover

    @traces(spec="004", story="US-007", reqs=["REQ-004-US-007-008"])
    def validate(self) -> ValidationResult:
        """Self-validate the packet.

        Performance target: < 5ms.
        """
        ...  # pragma: no cover


@runtime_checkable
class ISerializer(Protocol):
    """Interface for CXP packet serialization (Spec 004 Section 6).

    Supports JSON and binary (MessagePack) formats.
    """

    @traces(spec="004", story="US-006", reqs=["REQ-004-US-006-001"])
    def to_json(self, packet: IPacket, pretty: bool = False) -> str:
        """Serialize packet to JSON string.

        Performance target: < 10ms for 1KB payload.
        """
        ...  # pragma: no cover

    @traces(spec="004", story="US-006", reqs=["REQ-004-US-006-009"])
    def from_json(self, data: str) -> IPacket:
        """Deserialize packet from JSON string.

        Performance target: < 10ms.
        """
        ...  # pragma: no cover

    @traces(spec="004", story="US-006", reqs=["REQ-004-US-006-017"])
    def to_bytes(self, packet: IPacket) -> bytes:
        """Serialize packet to binary (MessagePack) format.

        Performance target: < 5ms for 1KB payload.
        """
        ...  # pragma: no cover

    @traces(spec="004", story="US-006", reqs=["REQ-004-US-006-021"])
    def from_bytes(self, data: bytes) -> IPacket:
        """Deserialize packet from binary format.

        Performance target: < 5ms.
        """
        ...  # pragma: no cover


@runtime_checkable
@traces(
    spec="004",
    story="US-007",
    reqs=["REQ-004-US-007-018"],
)
class IValidator(Protocol):
    """Interface for CXP packet validation (Spec 004 Section 7).

    Supports rule-based and schema-based validation.
    """

    @traces(spec="004", story="US-007", reqs=["REQ-004-US-007-008"])
    def validate(self, packet: IPacket) -> ValidationResult:
        """Validate packet against registered rules.

        Performance target: < 5ms.
        """
        ...  # pragma: no cover

    @traces(spec="004", story="US-007", reqs=["REQ-004-US-007-009"])
    def validate_schema(
        self, packet: IPacket, schema: dict[str, Any]
    ) -> ValidationResult:
        """Validate packet against a JSON schema."""
        ...  # pragma: no cover

    @traces(spec="004", story="US-007", reqs=["REQ-004-US-007-019"])
    def register_validator(self, field: str, validator: Callable[..., bool]) -> None:
        """Register a custom field validator."""
        ...  # pragma: no cover


@runtime_checkable
class IThreadManager(Protocol):
    """Interface for CXP thread management (Spec 004 Section 8).

    Manages conversation threads linking related packets.
    """

    @traces(spec="004", story="US-008", reqs=["REQ-004-US-008-016"])
    def create_thread(self, initial_packet: IPacket) -> str:
        """Create a new thread from an initial packet.

        Returns the thread_id.
        """
        ...  # pragma: no cover

    @traces(spec="004", story="US-008", reqs=["REQ-004-US-008-017"])
    def add_to_thread(self, thread_id: str, packet: IPacket) -> None:
        """Add a packet to an existing thread."""
        ...  # pragma: no cover

    @traces(spec="004", story="US-008", reqs=["REQ-004-US-008-005"])
    def get_thread_packets(self, thread_id: str) -> list[IPacket]:
        """Retrieve all packets in a thread in chronological order.

        Performance target: < 2ms.
        """
        ...  # pragma: no cover

    @traces(spec="004", story="US-008", reqs=["REQ-004-US-008-018"])
    def close_thread(self, thread_id: str) -> None:
        """Close a thread (no new packets accepted)."""
        ...  # pragma: no cover

    @traces(spec="004", story="US-008", reqs=["REQ-004-US-008-019"])
    def reopen_thread(self, thread_id: str) -> None:
        """Reopen a previously closed thread."""
        ...  # pragma: no cover


@runtime_checkable
class IDeliveryTracker(Protocol):
    """Interface for CXP delivery tracking (Spec 004 Section 4).

    Tracks delivery status transitions and history.
    """

    @traces(spec="004", story="US-004", reqs=["REQ-004-US-004-006"])
    def set_status(self, packet: IPacket, status: DeliveryStatus) -> None:
        """Update delivery status (validates transition against state machine)."""
        ...  # pragma: no cover

    @traces(spec="004", story="US-004", reqs=["REQ-004-US-004-018"])
    def get_history(self, packet: IPacket) -> list[tuple[DeliveryStatus, datetime]]:
        """Get the delivery status history for a packet."""
        ...  # pragma: no cover

    @traces(spec="004", story="US-004", reqs=["REQ-004-US-004-020"])
    def record_failure(self, packet: IPacket, reason: str) -> None:
        """Record a delivery failure with reason."""
        ...  # pragma: no cover

    @traces(spec="004", story="US-004", reqs=["REQ-004-US-004-021"])
    def increment_retry(self, packet: IPacket) -> int:
        """Increment retry count and return the new count."""
        ...  # pragma: no cover


@runtime_checkable
class IPrologBridge(Protocol):
    """Interface for CXP Prolog integration (Spec 004 Section 10).

    Bridges CXP packets to Prolog facts for logical reasoning.
    """

    @traces(spec="004", story="INTG", reqs=["REQ-004-INTG-001"])
    async def assert_packet(self, packet: IPacket, session_id: str) -> None:
        """Assert a packet as a Prolog fact (cxp_packet/12)."""
        ...  # pragma: no cover

    async def retract_packet(self, message_id: str) -> None:
        """Retract a packet fact by message_id."""
        ...  # pragma: no cover

    @traces(spec="004", story="INTG", reqs=["REQ-004-INTG-003"])
    async def query_by_thread(self, thread_id: str) -> list[IPacket]:
        """Query all packets in a thread via Prolog."""
        ...  # pragma: no cover

    @traces(spec="004", story="INTG", reqs=["REQ-004-INTG-005"])
    async def query_by_status(
        self,
        delivery_status: DeliveryStatus | None = None,
        work_status: WorkStatus | None = None,
    ) -> list[IPacket]:
        """Query packets by delivery and/or work status."""
        ...  # pragma: no cover