orchestrata_ultimate.gateway.handlers._session_events

Reference › orchestrata_ultimate › gateway › handlers › _session_events

SessionEventOps is the event-broadcast half of the session handler (SPEC-016 REQ-016-249). It holds the message processor, injected after construction with set_processor, and publishes session.events notifications through its broadcast. broadcast_opened sends an opened event with the session id, a timestamp, an active status and the purpose; broadcast_closed sends the matching close event with the session id and the number of migrated items. With no processor wired, both do nothing. It is used only by the session handler, next to _session_kg. See gateway.

At a glance

Class diagram

Dependency diagram

API

Classes

SessionEventOps

Decorators: traces(spec='016', story='US-013', reqs=['REQ-016-246', 'REQ-016-249', 'REQ-016-250']) · Line 25

Session lifecycle event broadcasts -- sole owner of the message processor.

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/gateway/handlers/_session_events.py

"""Session event-broadcast contract -- SessionEventOps.

Moved out of ``gateway/handlers/session.py`` as packet P3 of the
SessionHandler cohesion split (``plans/session-handler-split-design.md``
§3.3, ``plans/contracts/session-handler-split.mmd``). Sole owner of the
``_processor`` message-processor reference and the ``session.events``
broadcast payload shapes. Every guard, timestamp call, and payload shape
below is byte-identical to its origin in ``session.py``.
"""

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

from orchestrata_ultimate.core.base.traces import traces

logger = logging.getLogger(__name__)


@traces(
    spec="016",
    story="US-013",
    reqs=["REQ-016-246", "REQ-016-249", "REQ-016-250"],
)
class SessionEventOps:
    """Session lifecycle event broadcasts -- sole owner of the message processor.

    Moved out of ``gateway/handlers/session.py`` as packet P3 of the
    SessionHandler cohesion split (``plans/session-handler-split-design.md``
    §3.3, ``plans/contracts/session-handler-split.mmd``). Composes the
    injected message processor; every broadcast call below is byte-identical
    to its origin on ``SessionHandler``.
    """

    def __init__(self, processor: Any = None) -> None:
        """Wire the event ops to the message processor.

        Args:
            processor: Message processor for broadcast/push, or ``None``
                when not yet injected (see ``set_processor``).
        """
        self._processor = processor

    @traces(spec="016", story="US-013", reqs=["REQ-016-246"])
    def set_processor(self, processor: Any) -> None:
        """Inject the message processor after construction.

        Args:
            processor: MessageProcessor instance.
        """
        self._processor = processor

    @traces(spec="016", story="US-013", reqs=["REQ-016-249"])
    async def broadcast_opened(self, session: Any, purpose: str) -> None:
        """Broadcast a session.events "opened" notification, if wired."""
        if not self._processor:
            return
        await self._processor.broadcast(
            "session.events",
            {
                "session_id": session.session_id,
                "event_type": "opened",
                "timestamp": datetime.now(UTC).isoformat(),
                "details": {"status": "active", "purpose": purpose},
            },
        )

    @traces(spec="016", story="US-013", reqs=["REQ-016-250"])
    async def broadcast_closed(self, session_id: str, migrated: int) -> None:
        """Broadcast a session.events "closed" notification, if wired."""
        if not self._processor:
            return
        await self._processor.broadcast(
            "session.events",
            {
                "session_id": session_id,
                "event_type": "closed",
                "timestamp": datetime.now(UTC).isoformat(),
                "details": {"status": "closed", "memories_migrated": migrated},
            },
        )