orchestrata_ultimate.gateway.handlers.session

Reference › orchestrata_ultimate › gateway › handlers › session

SessionHandler serves the session.* RPCs: session.active, session.list, session.get, session.open, session.close, session.events and session.facts.subscribe. These are the work sessions tracked by the SessionController in the session package, not the WebSocket connections held by the connection manager. rpc creates it. See gateway.

The handler is a thin coordinator over two helpers. _session_kg owns every graph read and write, and _session_events owns the session.events broadcasts. _resolve_session_root takes the root from the constructor argument or the SESSION_ROOT setting and raises ValueError if neither is present; it never guesses from the working directory.

• handle_open requires a purpose, begins a session with the controller, persists it to the graph with persist_open, broadcasts the opened event and returns the session id with status active.
• handle_close requires a session_id, persists the summary and carry-over tasks with persist_close, then calls _migrate_close_memories, which asks the shared memory engine to move that session's hot memories to cold storage (SPEC-010) on a best-effort basis, and broadcasts the closed event.
• handle_active, handle_list and handle_get read sessions through the graph helper.
• handle_facts_subscribe subscribes the connection to the session.facts topic with the supplied filters, and handle_events serves session event queries.

At a glance

Class diagram

Dependency diagram

API

Classes

SessionHandler

Decorators: traces(spec='014', story='US-004', reqs=['REQ-014-051', 'REQ-014-053', 'REQ-014-054', 'REQ-014-056', 'REQ-014-059', 'RE…, traces(spec='016', story='US-013', reqs=['REQ-016-245', 'REQ-016-246', 'REQ-016-247', 'REQ-016-248', 'REQ-016-249', 'RE… · Line 87

Session lifecycle and subscription methods.

Functions

Source

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

"""Session RPC handlers -- list, get, open, close, events, facts.

Implements SPEC-014 FR-017 session management methods.
"""

import asyncio
import logging
import os
from collections.abc import Mapping
from pathlib import Path
from typing import Any, cast

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.gateway.handlers._session_events import SessionEventOps
from orchestrata_ultimate.gateway.handlers._session_kg import (
    SessionKgOps,
    _deserialize_session,
    build_active_query,
    build_get_query,
    build_list_query,
    shape_get_result,
)
from orchestrata_ultimate.gateway.pubsub import PubSubManager
from orchestrata_ultimate.session.session import SessionController

logger = logging.getLogger(__name__)

_SESSION_ROOT_ERROR = (
    "SessionHandler requires a session-persistence root: pass "
    "session_root=<path> to the constructor, or set the SESSION_ROOT "
    "environment variable. Neither the process cwd nor this package's own "
    "location is used as a fallback -- an implicit root previously caused "
    "session lifecycle git operations (commit/push) to land on whatever "
    "repository happened to be checked out at the process's cwd "
    "(fix/session-root, RUL-011)."
)


@traces(spec="016", story="US-013", reqs=["REQ-016-245"])
def _resolve_session_root(
    session_root: str | Path | None, env: Mapping[str, str] | None
) -> str | Path:
    """Resolve the explicit session root: *session_root*, else ``SESSION_ROOT``.

    Never falls back to the process cwd or this package's own location.

    Raises:
        ValueError: neither source supplies a root.
    """
    if session_root is not None:
        return session_root
    resolved_env = os.environ if env is None else env
    env_root = resolved_env.get("SESSION_ROOT")
    if env_root:
        return env_root
    raise ValueError(_SESSION_ROOT_ERROR)


@traces(
    spec="014",
    story="US-004",
    reqs=[
        "REQ-014-051",
        "REQ-014-053",
        "REQ-014-054",
        "REQ-014-056",
        "REQ-014-059",
        "REQ-014-061",
        "REQ-014-063",
        "REQ-014-064",
    ],
)
@traces(
    spec="016",
    story="US-013",
    reqs=[
        "REQ-016-245",
        "REQ-016-246",
        "REQ-016-247",
        "REQ-016-248",
        "REQ-016-249",
        "REQ-016-250",
        "REQ-016-251",
        "REQ-016-252",
    ],
)
class SessionHandler:
    """Session lifecycle and subscription methods."""

    @traces(spec="016", story="US-013", reqs=["REQ-016-245"])
    def __init__(
        self,
        pubsub: PubSubManager | None,
        processor: Any = None,
        session_root: str | Path | None = None,
        env: Mapping[str, str] | None = None,
        kg_ops: SessionKgOps | None = None,
        event_ops: SessionEventOps | None = None,
    ) -> None:
        """Wire handler to session subsystems.

        Args:
            pubsub: Pub/sub subscription manager, or ``None`` when pub/sub is
                unavailable (``handle_events``/``handle_facts_subscribe`` then
                report "PubSub unavailable" instead of erroring).
            processor: Message processor for broadcast/push.
            session_root: Explicit session-persistence root. Falls back to
                the ``SESSION_ROOT`` variable in *env* (or ``os.environ``
                when *env* is not given) when omitted. Never inferred from
                the process cwd or this package's own location -- see
                ``_SESSION_ROOT_ERROR`` (fix/session-root, RUL-011).
            env: Optional environment mapping consulted for ``SESSION_ROOT``
                when *session_root* is not given. Defaults to ``os.environ``.
            kg_ops: Optional pre-built ``SessionKgOps`` (DI seam); defaults
                to a real instance bound to this handler's own
                ``SessionController``.
            event_ops: Optional pre-built ``SessionEventOps`` (DI seam);
                defaults to a real instance bound to *processor*.

        Raises:
            ValueError: neither *session_root* nor ``SESSION_ROOT`` supplies
                a root.
        """
        self._pubsub = pubsub
        self._memory_engine: Any = None
        self._session_controller = SessionController(
            _resolve_session_root(session_root, env),
            explicit_root=True,
        )
        self._kg = kg_ops or SessionKgOps(self._session_controller)
        self._events = event_ops or SessionEventOps(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._events.set_processor(processor)

    @traces(spec="016", story="US-013", reqs=["REQ-016-246"])
    def set_memory_engine(self, engine: Any) -> None:
        """Inject the gateway's shared memory engine for cold migration.

        On session close the handler migrates that session's hot memories to
        cold storage (SPEC-010 FR-008). The engine is the same DI-registered
        singleton the memory RPC handler uses, so it sees exactly the entries
        stored during the session.

        Args:
            engine: The gateway's singleton ``MemoryEngine`` (or None to leave
                cold migration disabled).
        """
        self._memory_engine = engine

    @traces(spec="016", story="US-013", reqs=["REQ-016-246"])
    def set_graph_repository(self, graph_repository: Any) -> None:
        """Inject the operational-graph repository.

        Args:
            graph_repository: An AsyncGraphRepository instance.
        """
        self._kg.set_graph_repository(graph_repository)

    @traces(spec="016", story="US-013", reqs=["REQ-016-246"])
    async def handle_active(
        self,
        connection_id: str,
        params: dict[str, Any],
    ) -> dict[str, Any]:
        """Handle session.active -- get current active session info.

        Queries the graph for the most recent active session.

        Returns:
            Active session or null.
        """
        if not self._kg.available:
            return {"session": None, "error": "Graph unavailable"}

        try:
            query, query_params = build_active_query()
            results = await asyncio.wait_for(
                self._kg.execute(query, query_params),
                timeout=10.0,
            )
            if results:
                session = _deserialize_session(results[0])
                return {"session": session}
            return {"session": None}
        except TimeoutError:
            return {"session": None, "error": "Query timeout"}
        except Exception as exc:
            return {"session": None, "error": str(exc)}

    @traces(spec="016", story="US-013", reqs=["REQ-016-247"])
    @traces(spec="014", story="US-004", reqs=["REQ-014-051"])
    async def handle_list(
        self,
        connection_id: str,
        params: dict[str, Any],
    ) -> dict[str, Any]:
        """Handle session.list -- list sessions from the graph.

        Also serves SPEC-018 REQ-018-042 for Fireflies session linking.

        Args:
            connection_id: Caller connection ID.
            params: Optional status filter, limit, offset.

        Returns:
            Paginated list of sessions.
        """
        if not self._kg.available:
            return {"sessions": [], "error": "Graph unavailable"}

        try:
            query, query_params = build_list_query(params)
            results = await asyncio.wait_for(
                self._kg.execute(query, query_params),
                timeout=10.0,
            )
            sessions = [_deserialize_session(row) for row in results]
            return {"sessions": sessions}
        except TimeoutError:
            return {"sessions": [], "error": "Query timeout"}
        except Exception as exc:
            return {"sessions": [], "error": str(exc)}

    @traces(spec="016", story="US-013", reqs=["REQ-016-248"])
    @traces(
        spec="014",
        story="US-004",
        reqs=["REQ-014-053", "REQ-014-054"],
    )
    async def handle_get(
        self,
        connection_id: str,
        params: dict[str, Any],
    ) -> dict[str, Any]:
        """Handle session.get -- get session by ID from the graph.

        Args:
            connection_id: Caller connection ID.
            params: session_id (camelCase sessionId accepted).

        Returns:
            Session state dict with context and facts.
        """
        session_id = params.get("session_id") or params.get("sessionId", "")
        if not session_id:
            return {"error": "Missing session_id"}

        if not self._kg.available:
            return {"error": "Graph unavailable"}

        try:
            query, query_params = build_get_query(
                self._kg.session_key(session_id), str(session_id)
            )
            results = await asyncio.wait_for(
                self._kg.execute(query, query_params),
                timeout=10.0,
            )
            return {"session": shape_get_result(results)}
        except TimeoutError:
            return {"error": "Query timeout"}
        except Exception as exc:
            return {"error": str(exc)}

    @traces(spec="016", story="US-013", reqs=["REQ-016-249"])
    @traces(
        spec="014",
        story="US-004",
        reqs=["REQ-014-061", "REQ-014-057", "REQ-014-058"],
    )
    async def handle_open(
        self,
        connection_id: str,
        params: dict[str, Any],
    ) -> dict[str, Any]:
        """Handle session.open -- open a new session via SessionController.

        Creates a real session and persists it to the graph; see
        ``_session_kg.build_open_query`` for the session-254 root-connectivity
        ``HAS_SESSION`` MERGE this write performs.

        Args:
            connection_id: Caller connection ID.
            params: Session creation parameters (goals, portfolio, purpose).

        Returns:
            Created session info.
        """
        goals = params.get("goals", [])
        portfolio = params.get("portfolio", "")
        purpose = params.get("purpose", "")

        if not purpose:
            return {"error": "Missing purpose"}

        # Ensure context is initialized before beginning a session
        if not self._session_controller.ctx.file_exists():
            self._session_controller.ctx.initialize()

        # Create session via SessionController
        session = self._session_controller.begin(
            purpose=purpose,
            objectives=goals if isinstance(goals, list) else [goals],
        )

        await self._kg.persist_open(session, goals, portfolio, purpose)
        await self._events.broadcast_opened(session, purpose)

        return {"session_id": session.session_id, "status": "active"}

    @traces(spec="016", story="US-013", reqs=["REQ-016-250"])
    @traces(
        spec="014",
        story="US-004",
        reqs=["REQ-014-063", "REQ-014-064"],
    )
    async def handle_close(
        self,
        connection_id: str,
        params: dict[str, Any],
    ) -> dict[str, Any]:
        """Handle session.close -- close an active session via SessionController.

        Ends the session and updates Neo4j.

        Args:
            connection_id: Caller connection ID.
            params: Session close parameters (session_id, summary, carryover_tasks).

        Returns:
            Closed session info.
        """
        session_id = params.get("session_id", "")
        summary = params.get("summary", [])
        carryover = params.get("carryover_tasks", [])

        if not session_id:
            return {"error": "Missing session_id"}

        self._end_active_session(session_id, summary, carryover)
        await self._kg.persist_close(session_id, summary, carryover)
        migrated = await self._migrate_close_memories(session_id)
        await self._events.broadcast_closed(session_id, migrated)

        return {"closed": True, "session_id": session_id, "memories_migrated": migrated}

    @traces(spec="016", story="US-013", reqs=["REQ-016-250"])
    def _end_active_session(
        self, session_id: str, summary: Any, carryover: Any
    ) -> None:
        """End the SessionController session, if it's the active one."""
        active = self._session_controller.active_session
        if not (active and active.session_id == session_id):
            return
        achievements = summary if isinstance(summary, list) else [summary]
        upcoming = carryover if isinstance(carryover, list) else [carryover]
        self._session_controller.end(
            achievements=achievements,
            upcoming=upcoming,
        )

    @traces(spec="016", story="US-013", reqs=["REQ-016-250"])
    async def _migrate_close_memories(self, session_id: str) -> int:
        """Migrate this session's hot memories to cold storage (SPEC-010 FR-008).

        Best-effort and only when the shared engine was actually used this
        session (running): an unused engine has nothing to migrate and a
        missing hot store must not fail the close.
        """
        engine = self._memory_engine
        if engine is None or not getattr(engine, "_running", False):
            return 0
        try:
            return cast(int, await engine.migrate_to_cold(session_id))
        except Exception:
            return 0  # cold migration is best-effort

    @traces(spec="016", story="US-013", reqs=["REQ-016-251"])
    @traces(
        spec="014",
        story="US-004",
        reqs=["REQ-014-056"],
    )
    async def handle_events(
        self,
        connection_id: str,
        params: dict[str, Any],
    ) -> dict[str, Any]:
        """Handle session.events -- subscribe to session events.

        Args:
            connection_id: Caller connection ID.
            params: Subscription parameters (optional session_id filter).

        Returns:
            Subscription confirmation.
        """
        if not self._pubsub:
            return {"error": "PubSub unavailable"}

        sub_id = await self._pubsub.subscribe(connection_id, "session.events", params)
        return {
            "subscription_id": sub_id,
            "subscription": "session.events",
            "status": "subscribed",
            "filters": params,
        }

    @traces(spec="016", story="US-013", reqs=["REQ-016-252"])
    @traces(
        spec="014",
        story="US-004",
        reqs=["REQ-014-059"],
    )
    async def handle_facts_subscribe(
        self,
        connection_id: str,
        params: dict[str, Any],
    ) -> dict[str, Any]:
        """Handle session.facts.subscribe -- subscribe to session fact events.

        Args:
            connection_id: Caller connection ID.
            params: Fact subscription parameters (session_id, fact_types).

        Returns:
            Subscription confirmation.
        """
        if not self._pubsub:
            return {"error": "PubSub unavailable"}

        sub_id = await self._pubsub.subscribe(connection_id, "session.facts", params)
        return {
            "subscription_id": sub_id,
            "subscription": "session.facts",
            "status": "subscribed",
            "filters": params,
        }