orchestrata_ultimate.cxp.exceptions

Reference › orchestrata_ultimate › cxp › exceptions

exceptions is the exception hierarchy of the CXP package, rooted at CXPError. Nearly every other module of the package imports it, so callers can catch the root or a specific class. Each subclass takes a default message plus the context fields that matter for its failure, and an optional details dictionary. The classes group by area:

• Packet and format: InvalidPacketError (with a field name), SerializationError and DeserializationError (with the format).
• Validation: ValidationError (lists of errors and warnings), with FieldValidationError (field and constraint) and SchemaValidationError (schema errors) below it. validator and schema raise these.
• Delivery: DeliveryError (packet id and destination), DeliveryTimeoutError and MaxRetriesExceededError (attempts against the maximum), used by delivery_status.
• Routing: RoutingError and NoRouteError (the destination), used by router.
• Threads: ThreadError, ThreadNotFoundError and ThreadClosedError, used by threads and thread_store.
• Governance: PrologBridgeError (with the operation), raised by prolog_bridge.

See cxp.

At a glance

Class diagram

Dependency diagram

API

Classes

InvalidPacketError

Bases: CXPError · Decorators: traces(spec='004', story='US-003', reqs=['REQ-004-US-003-023']) · Line 53

Raised when a packet is structurally invalid.

SerializationError

Bases: CXPError · Decorators: traces(spec='004', story='US-003', reqs=['REQ-004-US-003-005', 'REQ-004-US-003-013']) · Line 77

Raised when packet serialization fails.

DeserializationError

Bases: CXPError · Line 91

Raised when packet deserialization fails.

ValidationError

Bases: CXPError · Line 108

Raised when packet validation fails.

FieldValidationError

Bases: ValidationError · Decorators: traces(spec='004', story='US-002', reqs=['REQ-004-US-002-017', 'REQ-004-US-002-018']) · Line 134

Raised when a specific field fails validation.

SchemaValidationError

Bases: ValidationError · Decorators: traces(spec='004', story='US-007', reqs=['REQ-004-US-007-012']) · Line 153

Raised when a packet fails JSON schema validation.

DeliveryError

Bases: CXPError · Decorators: traces(spec='004', story='INTG', reqs=['REQ-004-INTG-013']) · Line 169

Raised when packet delivery fails.

DeliveryTimeoutError

Bases: DeliveryError · Line 187

Raised when packet delivery times out.

MaxRetriesExceededError

Bases: DeliveryError · Decorators: traces(spec='004', story='US-004', reqs=['REQ-004-US-004-022']) · Line 205

Raised when delivery retries are exhausted.

RoutingError

Bases: CXPError · Line 226

Raised when packet routing fails.

NoRouteError

Bases: RoutingError · Line 241

Raised when no route exists for a destination.

ThreadError

Bases: CXPError · Line 261

Raised when thread operations fail.

ThreadNotFoundError

Bases: ThreadError · Line 276

Raised when a thread does not exist.

ThreadClosedError

Bases: ThreadError · Line 292

Raised when attempting to add to a closed thread.

PrologBridgeError

Bases: CXPError · Line 311

Raised when Prolog bridge operations fail.

Source

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

"""
CXP Protocol Exceptions.

Hierarchical exception classes for the Cross-agent eXchange Protocol.
All CXP errors inherit from CXPError for unified catch handling.

``CXPError`` and ``StateTransitionError`` themselves live in Layer 0
(``orchestrata_ultimate.core.packet.exceptions``) because ``CXPPacket``'s
state machines raise them and the packet vocabulary lives in ``core``
(K6). They are re-exported here unchanged — same class objects — so every
existing ``cxp.exceptions`` import path and ``except`` clause keeps working.
"""

# Spec: 004 - CXP Protocol

from typing import Any

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.core.packet.exceptions import (  # noqa: F401
    CXPError,
    StateTransitionError,
)

__all__ = [
    "CXPError",
    "InvalidPacketError",
    "SerializationError",
    "DeserializationError",
    "ValidationError",
    "FieldValidationError",
    "SchemaValidationError",
    "DeliveryError",
    "DeliveryTimeoutError",
    "MaxRetriesExceededError",
    "RoutingError",
    "NoRouteError",
    "ThreadError",
    "ThreadNotFoundError",
    "ThreadClosedError",
    "StateTransitionError",
    "PrologBridgeError",
]


# ---------- Packet errors ----------


@traces(
    spec="004",
    story="US-003",
    reqs=["REQ-004-US-003-023"],
)
class InvalidPacketError(CXPError):
    """Raised when a packet is structurally invalid."""

    def __init__(
        self,
        message: str = "Invalid packet",
        field: str | None = None,
        details: dict[str, Any] | None = None,
    ) -> None:
        d = details or {}
        if field:
            d["field"] = field
        super().__init__(message, code="INVALID_PACKET", details=d)
        self.field = field


# ---------- Serialization errors ----------


@traces(
    spec="004",
    story="US-003",
    reqs=["REQ-004-US-003-005", "REQ-004-US-003-013"],
)
class SerializationError(CXPError):
    """Raised when packet serialization fails."""

    def __init__(
        self,
        message: str = "Serialization failed",
        format: str = "json",
        details: dict[str, Any] | None = None,
    ) -> None:
        d = details or {}
        d["format"] = format
        super().__init__(message, code="SERIALIZATION_ERROR", details=d)


class DeserializationError(CXPError):
    """Raised when packet deserialization fails."""

    def __init__(
        self,
        message: str = "Deserialization failed",
        format: str = "json",
        details: dict[str, Any] | None = None,
    ) -> None:
        d = details or {}
        d["format"] = format
        super().__init__(message, code="DESERIALIZATION_ERROR", details=d)


# ---------- Validation errors ----------


class ValidationError(CXPError):
    """Raised when packet validation fails."""

    def __init__(
        self,
        message: str = "Validation failed",
        errors: list[str] | None = None,
        warnings: list[str] | None = None,
        details: dict[str, Any] | None = None,
    ) -> None:
        d = details or {}
        d["errors"] = errors or []
        d["warnings"] = warnings or []
        super().__init__(message, code="VALIDATION_ERROR", details=d)
        self.errors = errors or []
        self.warnings = warnings or []


@traces(
    spec="004",
    story="US-002",
    reqs=[
        "REQ-004-US-002-017",
        "REQ-004-US-002-018",
    ],
)
class FieldValidationError(ValidationError):
    """Raised when a specific field fails validation."""

    def __init__(
        self,
        field: str,
        message: str = "Field validation failed",
        constraint: str = "",
        details: dict[str, Any] | None = None,
    ) -> None:
        d = details or {}
        d["field"] = field
        d["constraint"] = constraint
        super().__init__(message, errors=[f"{field}: {message}"], details=d)
        self.field = field
        self.constraint = constraint


@traces(spec="004", story="US-007", reqs=["REQ-004-US-007-012"])
class SchemaValidationError(ValidationError):
    """Raised when a packet fails JSON schema validation."""

    def __init__(
        self,
        message: str = "Schema validation failed",
        schema_errors: list[str] | None = None,
        details: dict[str, Any] | None = None,
    ) -> None:
        super().__init__(message, errors=schema_errors, details=details)


# ---------- Delivery errors ----------


@traces(spec="004", story="INTG", reqs=["REQ-004-INTG-013"])
class DeliveryError(CXPError):
    """Raised when packet delivery fails."""

    def __init__(
        self,
        message: str = "Delivery failed",
        packet_id: str | None = None,
        destination: str | None = None,
        details: dict[str, Any] | None = None,
    ) -> None:
        d = details or {}
        if packet_id:
            d["packet_id"] = packet_id
        if destination:
            d["destination"] = destination
        super().__init__(message, code="DELIVERY_ERROR", details=d)


class DeliveryTimeoutError(DeliveryError):
    """Raised when packet delivery times out."""

    def __init__(
        self,
        message: str = "Delivery timed out",
        packet_id: str | None = None,
        timeout_seconds: int | None = None,
        details: dict[str, Any] | None = None,
    ) -> None:
        d = details or {}
        if timeout_seconds is not None:
            d["timeout_seconds"] = timeout_seconds
        super().__init__(message, packet_id=packet_id, details=d)
        self.code = "DELIVERY_TIMEOUT"


@traces(spec="004", story="US-004", reqs=["REQ-004-US-004-022"])
class MaxRetriesExceededError(DeliveryError):
    """Raised when delivery retries are exhausted."""

    def __init__(
        self,
        message: str = "Max delivery retries exceeded",
        packet_id: str | None = None,
        attempts: int = 0,
        max_attempts: int = 3,
        details: dict[str, Any] | None = None,
    ) -> None:
        d = details or {}
        d["attempts"] = attempts
        d["max_attempts"] = max_attempts
        super().__init__(message, packet_id=packet_id, details=d)
        self.code = "MAX_RETRIES_EXCEEDED"


# ---------- Routing errors ----------


class RoutingError(CXPError):
    """Raised when packet routing fails."""

    def __init__(
        self,
        message: str = "Routing failed",
        packet_id: str | None = None,
        details: dict[str, Any] | None = None,
    ) -> None:
        d = details or {}
        if packet_id:
            d["packet_id"] = packet_id
        super().__init__(message, code="ROUTING_ERROR", details=d)


class NoRouteError(RoutingError):
    """Raised when no route exists for a destination."""

    def __init__(
        self,
        destination: str,
        message: str = "",
        details: dict[str, Any] | None = None,
    ) -> None:
        d = details or {}
        d["destination"] = destination
        msg = message or f"No route to destination: {destination}"
        super().__init__(msg, details=d)
        self.code = "NO_ROUTE"
        self.destination = destination


# ---------- Thread errors ----------


class ThreadError(CXPError):
    """Raised when thread operations fail."""

    def __init__(
        self,
        message: str = "Thread operation failed",
        thread_id: str | None = None,
        details: dict[str, Any] | None = None,
    ) -> None:
        d = details or {}
        if thread_id:
            d["thread_id"] = thread_id
        super().__init__(message, code="THREAD_ERROR", details=d)


class ThreadNotFoundError(ThreadError):
    """Raised when a thread does not exist."""

    def __init__(
        self,
        thread_id: str,
        details: dict[str, Any] | None = None,
    ) -> None:
        super().__init__(
            message=f"Thread not found: {thread_id}",
            thread_id=thread_id,
            details=details,
        )
        self.code = "THREAD_NOT_FOUND"


class ThreadClosedError(ThreadError):
    """Raised when attempting to add to a closed thread."""

    def __init__(
        self,
        thread_id: str,
        details: dict[str, Any] | None = None,
    ) -> None:
        super().__init__(
            message=f"Thread is closed: {thread_id}",
            thread_id=thread_id,
            details=details,
        )
        self.code = "THREAD_CLOSED"


# ---------- Prolog bridge errors ----------


class PrologBridgeError(CXPError):
    """Raised when Prolog bridge operations fail."""

    def __init__(
        self,
        message: str = "Prolog bridge error",
        operation: str = "",
        details: dict[str, Any] | None = None,
    ) -> None:
        d = details or {}
        if operation:
            d["operation"] = operation
        super().__init__(message, code="PROLOG_BRIDGE_ERROR", details=d)