orchestrata_ultimate.cxp.session_adapter

Reference › orchestrata_ultimate › cxp › session_adapter

SessionAdapter scopes CXP packets to one session, since a session is the unit of work. bind_session attaches the adapter to a session id and raises CXPError (code SESSION_BIND_ERROR) for an empty id; unbind_session clears the id but keeps stored packets, and clear_session discards them. tag_packet writes the session id into the packet's metadata, and raises CXPError if the adapter is not bound. validate_session_isolation is true only when the packet's metadata session id matches the bound session. add_packet stores a packet and raises CXPError if the adapter is unbound or the packet belongs to a different session (code SESSION_ISOLATION_VIOLATION). get_session_packets returns a copy of the list, and get_session_stats summarises it. It depends only on core.packet and exceptions. See cxp.

At a glance

Class diagram

Dependency diagram

API

Classes

SessionAdapter

Decorators: traces(spec='004', story='INTG', reqs=['REQ-004-INTG-006']) · Line 30

Adapts CXP packets for session-scoped isolation.

Source

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

"""
CXP Session Adapter.

Adapts CXP packets for session context management, enforcing
session isolation per Article I of the Orchestrata governance spec.

Every packet in a session is tagged with the session_id and validated
against it. Packets from other sessions are rejected.

Author: Mohamed
"""

import logging
from typing import Any

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.core.packet import CXPPacket, DeliveryStatus
from orchestrata_ultimate.cxp.exceptions import CXPError

logger = logging.getLogger(__name__)

__all__ = ["SessionAdapter"]


@traces(
    spec="004",
    story="INTG",
    reqs=["REQ-004-INTG-006"],
)
class SessionAdapter:
    """Adapts CXP packets for session-scoped isolation.

    Sessions are the unit of work (Article I). Every packet flowing
    through the adapter is bound to exactly one session_id.  The adapter
    stores packets, tags them with the session, and rejects packets
    that belong to a different session.

    Usage::

        adapter = SessionAdapter()
        adapter.bind_session("session-001")
        tagged = adapter.tag_packet(packet)
        adapter.add_packet(tagged)
        assert adapter.validate_session_isolation(tagged)
    """

    def __init__(self, session_id: str = "") -> None:
        self._session_id: str = session_id
        self._packets: list[CXPPacket] = []

    # ------------------------------------------------------------------
    # Properties
    # ------------------------------------------------------------------

    @property
    def session_id(self) -> str:
        """Return the current session id (read-only)."""
        return self._session_id

    # ------------------------------------------------------------------
    # Binding
    # ------------------------------------------------------------------

    def bind_session(self, session_id: str) -> None:
        """Bind this adapter to a session.

        Args:
            session_id: The session identifier to bind to.

        Raises:
            CXPError: If *session_id* is empty.
        """
        if not session_id:
            raise CXPError(
                "Cannot bind to an empty session_id",
                code="SESSION_BIND_ERROR",
            )
        self._session_id = session_id

    def unbind_session(self) -> None:
        """Unbind from the current session.

        Clears the session_id but does **not** discard stored packets.
        Call :meth:`clear_session` first if packets must be discarded.
        """
        self._session_id = ""

    @traces(spec="004", story="INTG", reqs=["REQ-004-INTG-006"])
    def is_bound(self) -> bool:
        """Return ``True`` if the adapter is currently bound to a session."""
        return bool(self._session_id)

    # ------------------------------------------------------------------
    # Packet tagging & isolation
    # ------------------------------------------------------------------

    @traces(spec="004", story="INTG", reqs=["REQ-004-INTG-006"])
    def tag_packet(self, packet: CXPPacket) -> CXPPacket:
        """Tag *packet* with this adapter's session_id.

        The session_id is written into ``packet.metadata["session_id"]``.

        Args:
            packet: The CXP packet to tag.

        Returns:
            The same packet instance, now carrying the session_id in its
            metadata.

        Raises:
            CXPError: If the adapter is not bound to a session.
        """
        if not self.is_bound():
            raise CXPError(
                "Adapter is not bound to a session -- call bind_session() first",
                code="SESSION_NOT_BOUND",
            )
        packet.metadata["session_id"] = self._session_id
        return packet

    @traces(spec="004", story="INTG", reqs=["REQ-004-INTG-006"])
    def validate_session_isolation(self, packet: CXPPacket) -> bool:
        """Check that *packet* belongs to this adapter's session.

        Returns ``True`` when ``packet.metadata["session_id"]`` matches
        the adapter's current session_id.  Returns ``False`` otherwise
        (including when the adapter is unbound or the packet has no
        session_id tag).
        """
        if not self.is_bound():
            return False
        return packet.metadata.get("session_id") == self._session_id

    # ------------------------------------------------------------------
    # Packet storage
    # ------------------------------------------------------------------

    @traces(spec="004", story="INTG", reqs=["REQ-004-INTG-006"])
    def add_packet(self, packet: CXPPacket) -> None:
        """Add *packet* to this session's packet store.

        Args:
            packet: The packet to store.

        Raises:
            CXPError: If the adapter is not bound to a session.
            CXPError: If the packet does not belong to this session
                      (isolation violation).
        """
        if not self.is_bound():
            raise CXPError(
                "Adapter is not bound to a session -- call bind_session() first",
                code="SESSION_NOT_BOUND",
            )
        if not self.validate_session_isolation(packet):
            raise CXPError(
                f"Packet {packet.message_id} does not belong to session "
                f"{self._session_id!r}",
                code="SESSION_ISOLATION_VIOLATION",
            )
        self._packets.append(packet)

    def get_session_packets(self) -> list[CXPPacket]:
        """Return all packets stored in this session.

        Returns a shallow copy so the caller cannot mutate the internal
        list directly.
        """
        return list(self._packets)

    def clear_session(self) -> None:
        """Remove all packets from the session store."""
        self._packets.clear()

    # ------------------------------------------------------------------
    # Statistics
    # ------------------------------------------------------------------

    def get_session_stats(self) -> dict[str, Any]:
        """Return statistics about the packets in this session.

        Returns:
            A dict with keys:
                - ``session_id``: Current session identifier.
                - ``packet_count``: Total number of stored packets.
                - ``by_status``: Dict mapping delivery-status value to count.
                - ``by_type``: Dict mapping packet-type value to count.
        """
        by_status: dict[str, int] = {}
        by_type: dict[str, int] = {}

        for pkt in self._packets:
            # Delivery status
            status_val = (
                pkt.delivery_status.value
                if isinstance(pkt.delivery_status, DeliveryStatus)
                else str(pkt.delivery_status)
            )
            by_status[status_val] = by_status.get(status_val, 0) + 1

            # Packet type
            type_val = (
                pkt.packet_type.value
                if hasattr(pkt.packet_type, "value")
                else str(pkt.packet_type)
            )
            by_type[type_val] = by_type.get(type_val, 0) + 1

        return {
            "session_id": self._session_id,
            "packet_count": len(self._packets),
            "by_status": by_status,
            "by_type": by_type,
        }