orchestrata_ultimate.cxp.prolog_bridge

Reference › orchestrata_ultimate › cxp › prolog_bridge

PrologBridge connects CXP to the Prolog policy engine, so that governance rules written in Prolog can decide routing and so that packets can be queried as facts. router is its consumer, and it implements the IPrologBridge contract from interfaces. It uses the Prolog adapter in core.prolog_adapter, and failures raise PrologBridgeError. See cxp.

The constructor takes an optional engine and a rules path. When no Prolog backend (janus-swi) is installed, it logs a warning and runs in degraded mode: is_available is false, queries return empty results, and mutations raise PrologBridgeError. When a backend exists, it uses the given engine or obtains one, resolves the CXP rules file (_resolve_rules_path), consults it (_load_rules, which logs a warning if the file is missing or an error if loading fails) and declares the dynamic predicate (_ensure_dynamic).

• assert_packet converts a packet and session id to terms (_packet_to_terms) and asserts a cxp_packet/12 fact, escaping quotes and backslashes with _escape_prolog_string. retract_packet removes it by message id.
• query_by_thread and query_by_status read facts back and rebuild packets with _terms_to_packet.
• validate_route_via_prolog translates volk aliases to Prolog atoms and queries can_send_to/2; it returns false if the engine is unavailable, the rules are not loaded, or the query fails. validate_packet_via_prolog validates a whole packet and returns a flag with a reason.

At a glance

Class diagram

Dependency diagram

API

Constants

Classes

PrologBridge

Decorators: traces(spec='004', story='INTG', reqs=['REQ-004-INTG-001', 'REQ-004-INTG-003']) · Line 122

CXP-Prolog bridge implementing IPrologBridge.

Functions

Source

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

"""
CXP Prolog Bridge.

Implements IPrologBridge for CXP-Prolog integration per Spec 004 Article II.
Bridges CXP packets to/from Prolog facts for logical reasoning, route
validation, and packet validation via the rules defined in cxp.pl.

The bridge:
  - Asserts cxp_packet/12 facts into the Prolog knowledge base
  - Retracts packets by message_id
  - Queries packets by thread_id or status
  - Validates routing via can_send_to/2
  - Validates packets via validate_packet/2

Handles graceful degradation when janus-swi is not installed.

Author: Mohamed
"""

# Spec: 004 - CXP Protocol

import json
import logging
import os
from datetime import UTC, datetime
from typing import Any

from orchestrata_ultimate.core.packet import CXPPacket, DeliveryStatus, WorkStatus
from orchestrata_ultimate.core.paths import get_prolog_dir
from orchestrata_ultimate.core.tracing.traceability import traces
from orchestrata_ultimate.cxp.constants import PROTOCOL_VERSION
from orchestrata_ultimate.cxp.exceptions import PrologBridgeError

logger = logging.getLogger(__name__)

# --- Conditional Prolog import ---
# Use the queue-based PrologProxy (singleton worker thread) instead
# of direct Prolog — ensures all Prolog access is serialized
# through one thread and the Prolog engine is a true singleton.
try:
    from orchestrata_ultimate.orchestrator.prolog_worker import get_prolog_proxy

    PROLOG_AVAILABLE = True
except ImportError:
    try:
        from orchestrata_ultimate.core.prolog_adapter import (
            PROLOG_AVAILABLE,
            PrologEngine,
        )

        get_prolog_proxy = None  # type: ignore[assignment]
    except ImportError:
        PROLOG_AVAILABLE = False
        PrologEngine = None  # type: ignore[assignment,misc]
        get_prolog_proxy = None  # type: ignore[assignment]

__all__ = [
    "PrologBridge",
    "PROLOG_AVAILABLE",
]

# Mapping from Volk alias (U:Ox) to Prolog atom (ox)
_ALIAS_TO_ATOM: dict[str, str] = {
    "U:Ox": "ox",
    "U:PS": "ps",
    "U:Cd": "cd",
    "U:Ux": "ux",
    "U:PM": "pm",
    "U:Ar": "ar",
    "U:Qa": "qa",
    "U:Cx": "cx",
}

_ATOM_TO_ALIAS: dict[str, str] = {v: k for k, v in _ALIAS_TO_ATOM.items()}

# cxp_packet/12 field order:
#   cxp_packet(MessageId, FromAgent, ToAgent, ThreadId, Subject,
#              PayloadJson, Timestamp, Priority, DeliveryStatus,
#              WorkStatus, ProtocolVersion, SessionId)
_CXP_PACKET_ARITY = 12


def _alias_to_atom(alias: str) -> str:
    """Convert a Volk alias like 'U:Ox' to a Prolog atom like 'ox'.

    If the alias is already an atom (lowercase), return it as-is.
    """
    return _ALIAS_TO_ATOM.get(alias, alias.lower())


@traces(spec="004", story="INTG", reqs=["REQ-004-INTG-003", "REQ-004-INTG-005"])
def _atom_to_alias(atom: object) -> str:
    """Convert a Prolog atom like 'ox' to a Volk alias like 'U:Ox'.

    If no mapping exists, return the atom as-is.

    Accepts non-str input because janus-swi query results can bind
    non-string Prolog terms (e.g. numbers); the body coerces via
    ``str(atom)`` regardless, so the signature reflects what actually
    happens at runtime.
    """
    return _ATOM_TO_ALIAS.get(str(atom), str(atom))


@traces(spec="004", story="INTG", reqs=["REQ-004-INTG-001"])
def _escape_prolog_string(value: str) -> str:
    """Escape a string for safe embedding in a Prolog atom/string.

    Escapes single quotes and backslashes.
    """
    return value.replace("\\", "\\\\").replace("'", "\\'")


@traces(
    spec="004",
    story="INTG",
    reqs=[
        "REQ-004-INTG-001",
        "REQ-004-INTG-003",
    ],
)
class PrologBridge:
    """CXP-Prolog bridge implementing IPrologBridge.

    Bridges CXPPacket instances to Prolog cxp_packet/12 facts, enabling
    logical queries over the packet store. Integrates with cxp.pl rules
    for route validation (can_send_to/2) and packet validation
    (validate_packet/2).

    If janus-swi is not installed, all query methods return empty results
    and mutation methods raise PrologBridgeError.
    """

    def __init__(
        self,
        prolog_engine: Any = None,
        cxp_rules_path: str = "",
    ) -> None:
        """Initialize the Prolog bridge.

        Args:
            prolog_engine: Optional pre-configured PrologEngine (janus-swi) instance.
                           If None, a new engine is created (when janus-swi is available).
            cxp_rules_path: Path to cxp.pl rules file. If empty, the bridge
                            attempts to locate it relative to this file at
                            prolog/cxp.pl (relative to project root).
        """
        self._engine: Any = None
        self._dynamic_declared: bool = False
        self._rules_loaded: bool = False
        self._cxp_rules_path: str = cxp_rules_path

        if not PROLOG_AVAILABLE:
            logger.warning(
                "No Prolog backend available. PrologBridge operates in degraded mode. "
                "Install with: uv add janus-swi"
            )
            return

        if prolog_engine is not None:
            self._engine = prolog_engine
        elif get_prolog_proxy is not None:
            self._engine = get_prolog_proxy()
        else:
            # This branch is only reached when PROLOG_AVAILABLE came from the
            # core.prolog_adapter fallback import (see the module-level try/
            # except cascade above), where PrologEngine is always the real
            # class, never None. Pyright can't correlate that across the two
            # separately-narrowed globals, hence the suppression.
            #
            # Spelled as `# pyright: ignore`, not `# type: ignore`: the
            # suppression is FOR pyright (reportOptionalCall), and mypy sees no
            # error on this line, so the mypy form was reported as an unused
            # ignore. Two checkers, two comment dialects — using the wrong one
            # traded a pyright error for a mypy one.
            self._engine = PrologEngine()  # pyright: ignore[reportOptionalCall]
        self._cxp_rules_path = self._resolve_rules_path(cxp_rules_path)
        self._load_rules()
        self._ensure_dynamic()

    def _resolve_rules_path(self, cxp_rules_path: str) -> str:
        """Resolve the cxp.pl rules file path.

        Args:
            cxp_rules_path: Explicit path, or empty to auto-detect.

        Returns:
            Resolved path string (may be empty if not found).
        """
        if cxp_rules_path:
            return cxp_rules_path
        candidate = str(get_prolog_dir() / "cxp.pl")
        return candidate if os.path.isfile(candidate) else ""

    def _load_rules(self) -> None:
        """Load CXP Prolog rules from the resolved path, if available."""
        if not self._cxp_rules_path:
            return
        if not os.path.isfile(self._cxp_rules_path):
            logger.warning("CXP rules file not found: %s", self._cxp_rules_path)
            return
        try:
            self._engine.consult(self._cxp_rules_path)
            self._rules_loaded = True
            logger.info("Loaded CXP Prolog rules from %s", self._cxp_rules_path)
        except Exception as exc:
            logger.error(
                "Failed to load CXP rules from %s: %s", self._cxp_rules_path, exc
            )
            self._rules_loaded = False

    # ------------------------------------------------------------------
    # IPrologBridge interface methods
    # ------------------------------------------------------------------

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

        Args:
            packet: The CXPPacket to assert.
            session_id: Session identifier for context tracking.

        Raises:
            PrologBridgeError: If janus-swi is unavailable or assertion fails.
        """
        if not self.is_available():
            raise PrologBridgeError(
                "Prolog engine is not available (janus-swi not installed)",
                operation="assert_packet",
            )

        terms = self._packet_to_terms(packet, session_id)
        assertion = (
            f"assert(cxp_packet("
            f"'{_escape_prolog_string(terms[0])}', "
            f"'{_escape_prolog_string(terms[1])}', "
            f"'{_escape_prolog_string(terms[2])}', "
            f"'{_escape_prolog_string(terms[3])}', "
            f"'{_escape_prolog_string(terms[4])}', "
            f"'{_escape_prolog_string(terms[5])}', "
            f"'{_escape_prolog_string(terms[6])}', "
            f"{terms[7]}, "
            f"'{_escape_prolog_string(terms[8])}', "
            f"'{_escape_prolog_string(terms[9])}', "
            f"'{_escape_prolog_string(terms[10])}', "
            f"'{_escape_prolog_string(terms[11])}'"
            f"))"
        )

        try:
            list(self._engine.query(assertion))
            logger.debug("Asserted cxp_packet for message_id=%s", packet.message_id)
        except Exception as exc:
            raise PrologBridgeError(
                f"Failed to assert packet {packet.message_id}: {exc}",
                operation="assert_packet",
                details={"message_id": packet.message_id},
            ) from exc

    async def retract_packet(self, message_id: str) -> None:
        """Retract a cxp_packet/12 fact by message_id.

        Args:
            message_id: The message_id of the packet to retract.

        Raises:
            PrologBridgeError: If janus-swi is unavailable or retraction fails.
        """
        if not self.is_available():
            raise PrologBridgeError(
                "Prolog engine is not available (janus-swi not installed)",
                operation="retract_packet",
            )

        escaped_id = _escape_prolog_string(message_id)
        retraction = (
            f"retract(cxp_packet('{escaped_id}', _, _, _, _, _, _, _, _, _, _, _))"
        )

        try:
            list(self._engine.query(retraction))
            logger.debug("Retracted cxp_packet for message_id=%s", message_id)
        except Exception as exc:
            raise PrologBridgeError(
                f"Failed to retract packet {message_id}: {exc}",
                operation="retract_packet",
                details={"message_id": message_id},
            ) from exc

    @traces(spec="004", story="INTG", reqs=["REQ-004-INTG-003"])
    async def query_by_thread(self, thread_id: str) -> list[CXPPacket]:
        """Query all cxp_packet/12 facts matching a thread_id.

        Args:
            thread_id: The thread_id to query.

        Returns:
            List of CXPPacket instances in the thread. Empty list if
            janus-swi is unavailable.
        """
        if not self.is_available():
            return []

        escaped_tid = _escape_prolog_string(thread_id)
        query = (
            f"cxp_packet(MsgId, From, To, '{escaped_tid}', Subject, "
            f"Payload, Timestamp, Priority, DStatus, WStatus, Version, SessionId)"
        )

        packets: list[CXPPacket] = []
        try:
            results = list(self._engine.query(query))
            for result in results:
                terms = {
                    "message_id": str(result.get("MsgId", "")),
                    "from_agent": str(result.get("From", "")),
                    "to_agent": str(result.get("To", "")),
                    "thread_id": thread_id,
                    "subject": str(result.get("Subject", "")),
                    "payload_json": str(result.get("Payload", "{}")),
                    "timestamp_iso": str(result.get("Timestamp", "")),
                    "priority": result.get("Priority", 5),
                    "delivery_status": str(result.get("DStatus", "queued")),
                    "work_status": str(result.get("WStatus", "pending")),
                    "protocol_version": str(result.get("Version", PROTOCOL_VERSION)),
                    "session_id": str(result.get("SessionId", "")),
                }
                packet = self._terms_to_packet(terms)
                packets.append(packet)
        except Exception as exc:
            logger.error("query_by_thread(%s) failed: %s", thread_id, exc)

        return packets

    @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[CXPPacket]:
        """Query cxp_packet/12 facts by delivery and/or work status.

        Args:
            delivery_status: Filter by delivery status. None matches any.
            work_status: Filter by work status. None matches any.

        Returns:
            List of matching CXPPacket instances. Empty list if janus-swi
            is unavailable.
        """
        if not self.is_available():
            return []

        ds_term = f"'{delivery_status.value}'" if delivery_status else "DStatus"
        ws_term = f"'{work_status.value}'" if work_status else "WStatus"

        query = (
            f"cxp_packet(MsgId, From, To, ThreadId, Subject, "
            f"Payload, Timestamp, Priority, {ds_term}, {ws_term}, Version, SessionId)"
        )

        packets: list[CXPPacket] = []
        try:
            results = list(self._engine.query(query))
            for result in results:
                ds_val = (
                    delivery_status.value
                    if delivery_status
                    else str(result.get("DStatus", "queued"))
                )
                ws_val = (
                    work_status.value
                    if work_status
                    else str(result.get("WStatus", "pending"))
                )
                terms = {
                    "message_id": str(result.get("MsgId", "")),
                    "from_agent": str(result.get("From", "")),
                    "to_agent": str(result.get("To", "")),
                    "thread_id": str(result.get("ThreadId", "")),
                    "subject": str(result.get("Subject", "")),
                    "payload_json": str(result.get("Payload", "{}")),
                    "timestamp_iso": str(result.get("Timestamp", "")),
                    "priority": result.get("Priority", 5),
                    "delivery_status": ds_val,
                    "work_status": ws_val,
                    "protocol_version": str(result.get("Version", PROTOCOL_VERSION)),
                    "session_id": str(result.get("SessionId", "")),
                }
                packet = self._terms_to_packet(terms)
                packets.append(packet)
        except Exception as exc:
            logger.error("query_by_status failed: %s", exc)

        return packets

    # ------------------------------------------------------------------
    # Extended validation methods
    # ------------------------------------------------------------------

    @traces(spec="001", story="US-003", reqs=["REQ-001-US-003-001"])
    def validate_route_via_prolog(self, from_agent: str, to_agent: str) -> bool:
        """Validate that a route exists between two agents via can_send_to/2.

        Translates Volk aliases (U:Ox) to Prolog atoms (ox) before querying.

        Args:
            from_agent: Sender Volk alias or atom.
            to_agent: Receiver Volk alias or atom.

        Returns:
            True if the route is valid, False otherwise.
            Returns False if janus-swi is unavailable or rules are not loaded.
        """
        if not self.is_available() or not self._rules_loaded:
            return False

        from_atom = _alias_to_atom(from_agent)
        to_atom = _alias_to_atom(to_agent)

        query = f"can_send_to({from_atom}, {to_atom})"
        try:
            results = list(self._engine.query(query))
            return len(results) > 0
        except Exception as exc:
            logger.error(
                "validate_route_via_prolog(%s, %s) failed: %s",
                from_agent,
                to_agent,
                exc,
            )
            return False

    def validate_packet_via_prolog(self, packet: CXPPacket) -> tuple[bool, str]:
        """Validate a CXP packet via Prolog validate_packet/2.

        Constructs a packet/7 compound term from the CXPPacket and queries
        validate_packet(Packet, Result).

        Args:
            packet: The CXPPacket to validate.

        Returns:
            Tuple of (is_valid, reason). is_valid is True if Prolog returns
            'valid', False otherwise. reason contains the validation message.
            Returns (False, "Prolog engine not available") if janus-swi is
            unavailable.
        """
        if not self.is_available() or not self._rules_loaded:
            return (False, "Prolog engine not available")

        packet_term = self._build_packet_term(packet)
        query = f"validate_packet({packet_term}, Result)"

        try:
            results = list(self._engine.query(query))
            if not results:
                return (
                    False,
                    "Prolog validation returned no results",
                )  # pragma: no cover
            return self._interpret_validation_result(results[0])
        except Exception as exc:
            logger.error("validate_packet_via_prolog failed: %s", exc)
            return (False, f"Prolog query error: {exc}")

    def _build_packet_term(self, packet: CXPPacket) -> str:
        """Build the Prolog packet/7 compound term for validate_packet/2.

        Args:
            packet: The CXPPacket to represent.

        Returns:
            Prolog term string suitable for embedding in a query.
        """
        from_atom = _alias_to_atom(packet.from_agent)
        to_atom = _alias_to_atom(packet.to_agent)
        thread_atom = _escape_prolog_string(packet.thread_id)
        subject_atom = _escape_prolog_string(packet.subject)
        body_atom = _escape_prolog_string(
            json.dumps(packet.payload, default=str)[:200] if packet.payload else "empty"
        )
        ds_atom = packet.delivery_status.value
        ws_atom = packet.work_status.value
        return (
            f"packet({from_atom}, {to_atom}, '{thread_atom}', "
            f"'{subject_atom}', '{body_atom}', {ds_atom}, {ws_atom})"
        )

    def _interpret_validation_result(self, result: dict) -> tuple[bool, str]:
        """Interpret a single Prolog validate_packet/2 result binding.

        Args:
            result: A single result dict from janus-swi query, containing 'Result'.

        Returns:
            Tuple of (is_valid, reason).
        """
        result_str = str(result.get("Result", ""))

        if result_str == "valid":
            return (True, "valid")

        if "invalid" in result_str:
            reason = result_str
            if "(" in result_str and ")" in result_str:
                inner = result_str.split("(", 1)[1].rsplit(")", 1)[0]
                reason = inner.strip("'\"")
            return (False, reason)

        return (False, f"Unexpected result: {result_str}")  # pragma: no cover

    # ------------------------------------------------------------------
    # Availability check
    # ------------------------------------------------------------------

    def is_available(self) -> bool:
        """Check if the Prolog engine is available and functional.

        Returns:
            True if janus-swi is installed and the engine is initialized.
        """
        return PROLOG_AVAILABLE and self._engine is not None

    # ------------------------------------------------------------------
    # Internal helpers
    # ------------------------------------------------------------------

    @traces(spec="004", story="INTG", reqs=["REQ-004-INTG-001"])
    def _packet_to_terms(self, packet: CXPPacket, session_id: str) -> list:
        """Convert a CXPPacket to a list of 12 Prolog term values.

        Field order matches cxp_packet/12:
            [message_id, from_agent, to_agent, thread_id, subject,
             payload_json, timestamp_iso, priority, delivery_status,
             work_status, protocol_version, session_id]

        Args:
            packet: The CXPPacket to convert.
            session_id: Session identifier.

        Returns:
            List of 12 string/int values suitable for Prolog assertion.
        """
        # Serialize payload to JSON string for Prolog storage
        if packet.payload is not None:
            try:
                payload_json = json.dumps(packet.payload, default=str)
            except TypeError, ValueError:
                payload_json = "{}"
        else:
            payload_json = "{}"

        # Serialize timestamp to ISO format
        ts = packet.timestamp
        if isinstance(ts, datetime):
            timestamp_iso = ts.isoformat()
        else:
            timestamp_iso = str(ts) if ts else datetime.now(UTC).isoformat()

        return [
            packet.message_id,  # 1: MessageId
            _alias_to_atom(packet.from_agent),  # 2: FromAgent (Prolog atom)
            _alias_to_atom(packet.to_agent),  # 3: ToAgent (Prolog atom)
            packet.thread_id,  # 4: ThreadId
            packet.subject,  # 5: Subject
            payload_json,  # 6: PayloadJson
            timestamp_iso,  # 7: Timestamp (ISO string)
            int(packet.priority),  # 8: Priority (integer)
            packet.delivery_status.value,  # 9: DeliveryStatus
            packet.work_status.value,  # 10: WorkStatus
            packet.protocol_version,  # 11: ProtocolVersion
            session_id,  # 12: SessionId
        ]

    @traces(spec="004", story="INTG", reqs=["REQ-004-INTG-003", "REQ-004-INTG-005"])
    def _terms_to_packet(self, terms: dict) -> CXPPacket:
        """Convert Prolog query result terms back to a CXPPacket.

        Args:
            terms: Dictionary with keys matching cxp_packet/12 field names:
                   message_id, from_agent, to_agent, thread_id, subject,
                   payload_json, timestamp_iso, priority, delivery_status,
                   work_status, protocol_version, session_id.

        Returns:
            Reconstructed CXPPacket instance.
        """
        # Parse payload from JSON
        payload_json = terms.get("payload_json", "{}")
        try:
            payload = json.loads(payload_json)
        except json.JSONDecodeError, TypeError:
            payload = {"raw": payload_json}

        # Parse timestamp
        timestamp_iso = terms.get("timestamp_iso", "")
        timestamp: datetime | None = None
        if timestamp_iso:
            try:
                timestamp = datetime.fromisoformat(timestamp_iso.replace("Z", "+00:00"))
            except ValueError, TypeError:
                timestamp = None
        if timestamp is None:
            timestamp = datetime.now(UTC)

        # Parse priority
        try:
            priority = int(terms.get("priority", 5))
        except ValueError, TypeError:
            priority = 5

        # Map Prolog atoms back to Volk aliases
        from_agent = _atom_to_alias(terms.get("from_agent", ""))
        to_agent = _atom_to_alias(terms.get("to_agent", ""))

        # Parse enums
        try:
            ds = DeliveryStatus(terms.get("delivery_status", "queued"))
        except ValueError:
            ds = DeliveryStatus.QUEUED

        try:
            ws = WorkStatus(terms.get("work_status", "pending"))
        except ValueError:
            ws = WorkStatus.PENDING

        return CXPPacket(
            message_id=terms.get("message_id", ""),
            from_agent=from_agent,
            to_agent=to_agent,
            thread_id=terms.get("thread_id", ""),
            subject=terms.get("subject", ""),
            payload=payload,
            timestamp=timestamp,
            priority=priority,
            delivery_status=ds,
            work_status=ws,
            protocol_version=terms.get("protocol_version", PROTOCOL_VERSION),
        )

    def _ensure_dynamic(self) -> None:
        """Ensure cxp_packet/12 is declared as a dynamic predicate.

        The cxp.pl file does not declare cxp_packet/12, so the bridge
        must declare it before any assert/retract operations.
        """
        if not self.is_available():
            return

        if self._dynamic_declared:
            return

        try:
            list(self._engine.query(f"dynamic(cxp_packet/{_CXP_PACKET_ARITY})"))
            self._dynamic_declared = True
            logger.debug("Declared cxp_packet/%d as dynamic", _CXP_PACKET_ARITY)
        except Exception as exc:
            # Some SWI-Prolog versions need :- dynamic declaration differently
            logger.warning(
                "dynamic/1 declaration failed (%s), trying assertz workaround",
                exc,
            )
            try:
                # Assert and immediately retract a dummy fact to implicitly
                # create the dynamic predicate
                dummy = (
                    "cxp_packet('__init__','_','_','_','_','_','_',0,'_','_','_','_')"
                )
                list(self._engine.query(f"assert({dummy})"))  # pragma: no cover
                list(self._engine.query(f"retract({dummy})"))  # pragma: no cover
                self._dynamic_declared = True  # pragma: no cover
                logger.debug(  # pragma: no cover
                    "Declared cxp_packet/%d via assert/retract workaround",
                    _CXP_PACKET_ARITY,
                )
            except Exception as exc2:
                logger.error(
                    "Failed to declare cxp_packet/%d as dynamic: %s",
                    _CXP_PACKET_ARITY,
                    exc2,
                )