orchestrata_ultimate.session.event_emitting_session_controller

Reference › orchestrata_ultimate › session › event_emitting_session_controller

EventEmittingSessionController wraps a SessionController so that session lifecycle events reach the kernel's event bus (spec 016). It takes the controller and an optional event bus. begin, end and log_work call the wrapped controller and then emit session.opened, session.closed and session.work.logged, with the sequence number, purpose and objectives in the payload of the first. Emission is fire-and-forget on a daemon thread, so a synchronous caller is not blocked. With no bus the wrapper is a transparent pass-through, and __getattr__ forwards every other attribute to the wrapped controller.

At a glance

Class diagram

Dependency diagram

API

Classes

EventEmittingSessionController

Decorators: traces(spec='016', story='US-013', reqs=['REQ-016-251']) · Line 30

SessionController wrapper that emits session lifecycle events.

Functions

Source

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

# mypy: ignore-errors
"""Event-emitting wrapper for SessionController lifecycle events."""

import asyncio
import logging
import threading
from typing import Any

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.core.protocols import IEventBus, SystemEvent
from orchestrata_ultimate.session.session import SessionController, SessionRecord

logger = logging.getLogger(__name__)


@traces(spec="016", story="US-013", reqs=["REQ-016-251"])
def _fire_and_forget(event_bus: IEventBus, event: SystemEvent) -> None:
    """Emit event in a daemon thread — sync callers do not block on it."""

    def _run() -> None:
        try:
            asyncio.run(event_bus.emit(event))
        except Exception:
            logger.exception("Failed to emit %s", event.type)

    threading.Thread(target=_run, daemon=True).start()


@traces(spec="016", story="US-013", reqs=["REQ-016-251"])
class EventEmittingSessionController:
    """SessionController wrapper that emits session lifecycle events.

    Wraps a SessionController instance and emits events to an optional
    IEventBus on begin, end, and log_work operations. Emission is
    fire-and-forget (daemon thread) so sync callers are not blocked.
    When event_bus is None the wrapper is a transparent pass-through.

    Events emitted:
        session.opened  — on begin()
        session.closed  — on end()
        session.work.logged — on log_work()
    """

    def __init__(
        self,
        session_controller: SessionController,
        event_bus: IEventBus | None = None,
    ) -> None:
        """Initialise the wrapper.

        Args:
            session_controller: The SessionController instance to wrap.
            event_bus: Optional event bus for emission. No events are emitted
                if None.
        """
        self._session_controller = session_controller
        self._event_bus = event_bus

    def __getattr__(self, name: str) -> Any:
        """Delegate all other attribute access to the wrapped SessionController."""
        return getattr(self._session_controller, name)

    @traces(spec="016", story="US-013", reqs=["REQ-016-251"])
    def begin(
        self,
        purpose: str,
        objectives: list[str] | None = None,
    ) -> SessionRecord:
        """Begin a new session and emit session.opened.

        Args:
            purpose: Session focus/purpose.
            objectives: Optional list of goals.

        Returns:
            Created SessionRecord.
        """
        result = self._session_controller.begin(purpose=purpose, objectives=objectives)
        if self._event_bus is not None:
            _fire_and_forget(
                self._event_bus,
                SystemEvent(
                    type="session.opened",
                    source="session_controller",
                    payload={
                        "seq": result.seq,
                        "purpose": purpose,
                        "objectives": objectives or [],
                    },
                ),
            )
        return result

    @traces(spec="016", story="US-013", reqs=["REQ-016-251"])
    def end(
        self,
        achievements: list[str] | None = None,
        upcoming: list[str] | None = None,
        *,
        skip_goal_check: bool = False,
    ) -> SessionRecord | None:
        """End the current session and emit session.closed.

        Args:
            achievements: Optional list of achievements.
            upcoming: Optional list of upcoming tasks.
            skip_goal_check: Passed through to the wrapped controller's
                SPEC-023 Stop gate escape hatch.

        Returns:
            Closed SessionRecord, or None if there was no active session.
        """
        result = self._session_controller.end(
            achievements=achievements,
            upcoming=upcoming,
            skip_goal_check=skip_goal_check,
        )
        if result is not None and self._event_bus is not None:
            _fire_and_forget(
                self._event_bus,
                SystemEvent(
                    type="session.closed",
                    source="session_controller",
                    payload={
                        "seq": result.seq,
                        "achievements": achievements or [],
                        "upcoming": upcoming or [],
                    },
                ),
            )
        return result

    @traces(spec="016", story="US-013", reqs=["REQ-016-251"])
    def log_work(self, desc: str, details: list[str] | None = None) -> None:
        """Log work and emit session.work.logged.

        Args:
            desc: Description of the work done.
            details: Optional bullet-point details.
        """
        self._session_controller.log_work(desc, details)
        if self._event_bus is not None:
            active = self._session_controller.active_session
            if active is not None:
                _fire_and_forget(
                    self._event_bus,
                    SystemEvent(
                        type="session.work.logged",
                        source="session_controller",
                        payload={"seq": active.seq, "desc": desc},
                    ),
                )