orchestrata_ultimate.gateway.handlers._session_kg

Reference › orchestrata_ultimate › gateway › handlers › _session_kg

_session_kg holds the graph side of the session handler. It was split out of the session handler, which is its only importer, so the handler keeps RPC logic and this module keeps the Cypher.

The module-level functions are pure query builders that return a query string and its parameters: build_active_query, build_list_query, build_get_query, build_open_query and build_close_query. shape_get_result and _deserialize_session turn rows into session dictionaries. build_open_query merges a Session node and a HAS_SESSION edge from the Project hub node, so a new session is never left unreachable from the project root.

SessionKgOps holds the stateful part and is the sole owner of the graph repository, which set_graph_repository injects. project_slug derives the project name from the session controller's root at call time, and session_key builds the project-qualified merge key of the form project, session, sequence, because a bare sequence number would collide across projects. execute runs a query in a managed read or write transaction and converts values with utils. persist_open and persist_close write the session lifecycle to the graph. See gateway.

At a glance

Class diagram

Dependency diagram

API

Classes

SessionKgOps

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

Session KG identity + stateful graph I/O -- sole owner of the graph repository.

Functions

Source

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

"""Session KG query-building contract -- pure functions + stateful I/O.

Moved out of ``gateway/handlers/session.py`` as packets P1 (pure Cypher
builders) and P2 (``SessionKgOps``) of the SessionHandler cohesion split
(``plans/session-handler-split-design.md`` §3.1/§3.2,
``plans/contracts/session-handler-split.mmd``). Every Cypher text below
is semantically identical (re-indented) to its origin in ``session.py``;
the only change for the
module functions is that project/session-key resolution (formerly
``self._kg_session_key`` / ``self._project_slug`` on ``SessionHandler``) is
now passed in explicitly by the caller, since these are stateless module
functions. This module owns the ``coalesce(s.id, s.session_id)`` schema-drift
contract in one place.
"""

import asyncio
import json
import logging
from pathlib import Path
from typing import Any

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.gateway.handlers.utils import _neo4j_value_to_dict
from orchestrata_ultimate.session.session import SessionController

logger = logging.getLogger(__name__)


@traces(spec="016", story="US-013", reqs=["REQ-016-246"])
def build_active_query() -> tuple[str, dict[str, Any]]:
    """Build the session.active Cypher query + params (extracted inline literal)."""
    query = """
    MATCH (s:Session)
    WHERE s.status = "active"
    RETURN coalesce(s.id, s.session_id) as session_id,
           s.status as status,
           s.goals as goals,
           s.portfolio as portfolio,
           s.purpose as purpose,
           s.start_time as start_time,
           s.duration_seconds as duration_seconds
    ORDER BY s.start_time DESC
    LIMIT 1
    """
    return query, {}


@traces(spec="016", story="US-013", reqs=["REQ-016-247"])
def build_list_query(params: dict[str, Any]) -> tuple[str, dict[str, Any]]:
    """Build the session.list Cypher query + params.

    Applies pagination (limit/offset, capped at 1000) and an optional
    ``status`` filter -- the query-building half of ``handle_list``.
    """
    limit = min(int(params.get("limit", 50)), 1000)
    offset = max(int(params.get("offset", 0)), 0)
    status_filter = params.get("status")

    query = """
    MATCH (s:Session)
    WHERE s.status IS NOT NULL
    """
    query_params: dict[str, Any] = {"limit": limit, "offset": offset}
    if status_filter:
        query += " AND s.status = $status"
        query_params["status"] = status_filter
    query += """
    RETURN coalesce(s.id, s.session_id) as session_id,
           s.status as status,
           s.goals as goals,
           s.portfolio as portfolio,
           s.purpose as purpose,
           s.start_time as start_time,
           s.duration_seconds as duration_seconds
    ORDER BY s.start_time DESC
    SKIP $offset
    LIMIT $limit
    """
    return query, query_params


@traces(spec="016", story="US-013", reqs=["REQ-016-248"])
def build_get_query(kg_key: str, raw_id: str) -> tuple[str, dict[str, Any]]:
    """Build the session.get Cypher query + params (id + fact/context lookup).

    Matches BOTH id shapes the session API itself emits: the canonical
    project-qualified merge key AND the raw id as provided.
    ``handle_list`` returns ``coalesce(s.id, s.session_id)`` — on nodes
    where ``s.id`` is a bare non-numeric id (the append-only Session
    writers), qualifying it produces a key no node carries, so a
    list→get round-trip returned ``{"session": None}`` (reviewer finding,
    fixed 2026-08-05). *kg_key* / *raw_id* are the two id shapes the caller
    already resolved (project-qualified merge key, raw id as given).
    """
    query = """
    MATCH (s:Session)
    WHERE (s.session_id = $session_key
           OR s.session_id = $raw_id
           OR s.id = $raw_id)
      AND s.status IS NOT NULL
    OPTIONAL MATCH (s)-[:HAS_FACT]->(f:Fact)
    RETURN coalesce(s.id, s.session_id) as session_id,
           s.status as status,
           s.goals as goals,
           s.portfolio as portfolio,
           s.purpose as purpose,
           s.start_time as start_time,
           s.duration_seconds as duration_seconds,
           s.context as context,
           collect(DISTINCT f{.*}) as facts,
           CASE WHEN s.session_id = $session_key THEN 0
                WHEN s.session_id = $raw_id THEN 1
                ELSE 2 END as pref
    ORDER BY pref
    LIMIT 1
    """
    return query, {
        "session_key": kg_key,
        "raw_id": raw_id,
    }


@traces(spec="016", story="US-013", reqs=["REQ-016-248"])
def shape_get_result(results: list[dict[str, Any]]) -> dict[str, Any] | None:
    """Shape the session.get Cypher rows into the response session dict."""
    if not results:
        return None
    row = dict(results[0])
    # Query-internal ordering column (deterministic id-shape preference),
    # not part of the session response surface.
    row.pop("pref", None)
    session = _deserialize_session(row)
    session["facts"] = row.get("facts", [])
    session["context"] = row.get("context")
    return session


@traces(spec="016", story="US-013", reqs=["REQ-016-249"])
def build_open_query(
    kg_key: str, project: str, session: Any, goals: Any, portfolio: str, purpose: str
) -> tuple[str, dict[str, Any]]:
    """Graph MERGE query + params for session.open.

    Session-254 root-connectivity ruling: MERGEs a ``HAS_SESSION`` edge
    from the ``:Project`` hub so the session is never orphaned (54 live
    Sessions were found unreachable by ``check_kg_connectivity.py``
    before this fix). *kg_key* / *project* are the same project-qualified
    merge key and project slug the caller resolves for itself.
    """
    query = """
    MERGE (s:Session {session_id: $session_id})
    SET s.id = $bare_id,
        s.seq = $seq,
        s.status = "active",
        s.goals = $goals,
        s.portfolio = $portfolio,
        s.purpose = $purpose,
        s.start_time = $start_time,
        s.duration_seconds = null,
        s.created_at = coalesce(s.created_at, datetime())
    WITH s
    MERGE (p:Project {slug: $project})
    MERGE (p)-[:HAS_SESSION]->(s)
    """
    query_params = {
        "session_id": kg_key,
        "bare_id": session.session_id,
        "seq": session.seq,
        "goals": json.dumps(goals) if goals else "[]",
        "portfolio": portfolio,
        "purpose": purpose,
        "start_time": session.when.isoformat(),
        "project": project,
    }
    return query, query_params


@traces(spec="016", story="US-013", reqs=["REQ-016-250"])
def build_close_query(
    kg_key: str, raw_id: str, summary: Any, carryover: Any
) -> tuple[str, dict[str, Any]]:
    """Build the session.close Cypher query + params (extracted inline literal).

    Matches the SAME three-shape identity surface ``build_get_query`` reads
    (project-qualified merge key, raw id via ``s.session_id``, raw id via
    ``s.id``), with the same deterministic id-shape preference. A raw-id-only
    Session node (append-only writer shape: bare ``s.id``, no qualified
    ``session_id``) is readable by ``session.get`` but was invisible to the
    old ``{session_id: $session_id}`` MATCH here -- SET touched nothing and
    the handler still replied ``{"closed": true}`` (backlog #399). *kg_key* /
    *raw_id* are the two id shapes the caller already resolved, matching
    ``build_get_query``'s contract.
    """
    query = """
    MATCH (s:Session)
    WHERE (s.session_id = $session_key
           OR s.session_id = $raw_id
           OR s.id = $raw_id)
      AND s.status IS NOT NULL
    WITH s, CASE WHEN s.session_id = $session_key THEN 0
                 WHEN s.session_id = $raw_id THEN 1
                 ELSE 2 END AS pref
    ORDER BY pref
    LIMIT 1
    SET s.status = "closed",
        s.closed_at = datetime(),
        s.summary = $summary,
        s.carryover = $carryover
    RETURN s
    """
    return query, {
        "session_key": kg_key,
        "raw_id": raw_id,
        "summary": json.dumps(summary) if summary else "[]",
        "carryover": json.dumps(carryover) if carryover else "[]",
    }


@traces(
    spec="016",
    story="US-013",
    reqs=["REQ-016-246", "REQ-016-247", "REQ-016-248"],
)
def _deserialize_session(row: dict[str, Any]) -> dict[str, Any]:
    """Deserialize a session record from the graph into a gateway dict."""
    session: dict[str, Any] = dict(row)
    goals = session.get("goals")
    if isinstance(goals, str):
        try:
            session["goals"] = json.loads(goals)
        except json.JSONDecodeError:
            session["goals"] = []
    elif goals is None:
        session["goals"] = []

    # Clean up None values for cleaner JSON
    for key in list(session.keys()):
        if session[key] is None:
            del session[key]

    return session


@traces(
    spec="016",
    story="US-013",
    reqs=["REQ-016-245", "REQ-016-246", "REQ-016-249", "REQ-016-250"],
)
class SessionKgOps:
    """Session KG identity + stateful graph I/O -- sole owner of the graph repository.

    Moved out of ``gateway/handlers/session.py`` as packet P2 of the
    SessionHandler cohesion split (``plans/session-handler-split-design.md``
    §3.2, ``plans/contracts/session-handler-split.mmd``). Composes the live
    ``SessionController`` (root read at call time only -- RUL-003, never
    cached) and the injected ``AsyncGraphRepository``. Every Cypher
    execution path, timeout, and log message below is byte-identical to its
    origin on ``SessionHandler``.
    """

    def __init__(
        self,
        session_controller: SessionController,
        graph_repository: Any = None,
    ) -> None:
        """Wire the KG ops to the live session controller + graph repository.

        Args:
            session_controller: The handler's ``SessionController`` -- only
                ``.root`` is read, and only at call time (RUL-003).
            graph_repository: An AsyncGraphRepository instance, or ``None``
                when not yet injected (see ``set_graph_repository``).
        """
        self._session_controller = session_controller
        self._graph_repository = graph_repository

    @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._graph_repository = graph_repository

    @property
    @traces(spec="016", story="US-013", reqs=["REQ-016-246"])
    def available(self) -> bool:
        """Whether a graph repository has been injected."""
        return bool(self._graph_repository)

    @traces(spec="016", story="US-013", reqs=["REQ-016-245"])
    def project_slug(self) -> str:
        """The project identity this handler's session root resolves to.

        Read at call time (RUL-003) from the live ``_session_controller``
        root -- never cached -- so the constructor's ``session_root=`` seam
        (or ``SESSION_ROOT`` env) is always the single source consulted.
        Shared by ``session_key`` (the Session merge key) and
        ``handle_open``'s ``:Project`` hub MERGE, so there is exactly one
        source of truth for "what project does this session belong to".
        """
        return Path(self._session_controller.root).resolve().name

    @traces(spec="016", story="US-013", reqs=["REQ-016-245"])
    def session_key(self, session_id: str | int) -> str:
        """Canonical project-qualified Session merge key (D4 shape).

        The public RPC surface keeps the bare zero-padded id; the KG node is
        keyed on ``{project}:session:{seq}`` (CANONICAL_MERGE_KEYS) because a
        bare seq collides across projects under the ``session_session_id``
        uniqueness constraint. Already-qualified ids pass through unchanged.
        """
        text = str(session_id)
        if ":" in text:
            return text
        project = self.project_slug()
        try:
            return f"{project}:session:{int(text)}"
        except ValueError:
            return f"{project}:session:{text}"

    @traces(spec="016", story="US-013", reqs=["REQ-016-246"])
    async def execute(
        self,
        query: str,
        params: dict | None = None,
        *,
        write: bool = False,
    ) -> list[dict[str, Any]]:
        """Execute a Cypher query against the graph.

        Args:
            query: Cypher string.
            params: Query parameters.
            write: True to run in a managed write transaction
                (CREATE/SET/...); False (default) for a managed read
                transaction.

        Returns:
            List of record dicts.
        """
        exec_fn = (
            self._graph_repository.execute_write
            if write
            else self._graph_repository.execute_read
        )
        rows = await exec_fn(query, params or {})
        return [
            {key: _neo4j_value_to_dict(val) for key, val in row.items()} for row in rows
        ]

    @traces(spec="016", story="US-013", reqs=["REQ-016-249"])
    async def persist_open(
        self, session: Any, goals: Any, portfolio: str, purpose: str
    ) -> None:
        """Persist a newly-opened session to the graph (best-effort, never silent).

        A swallowed constraint violation here is how cross-project
        collisions used to hide (see ``build_open_query``).
        """
        if not self.available:
            return

        query, query_params = build_open_query(
            self.session_key(session.seq),
            self.project_slug(),
            session,
            goals,
            portfolio,
            purpose,
        )
        try:
            await asyncio.wait_for(
                self.execute(query, query_params, write=True),
                timeout=10.0,
            )
        except Exception as exc:
            # Best-effort, but never silent: a swallowed constraint
            # violation here is how cross-project collisions hid.
            logger.warning("session.open KG write failed: %s", exc)

    @traces(spec="016", story="US-013", reqs=["REQ-016-250"])
    async def persist_close(
        self, session_id: str, summary: Any, carryover: Any
    ) -> None:
        """Update the graph session node to closed (best-effort, never silent)."""
        if not self.available:
            return
        query, query_params = build_close_query(
            self.session_key(session_id), str(session_id), summary, carryover
        )
        try:
            await asyncio.wait_for(
                self.execute(query, query_params, write=True),
                timeout=10.0,
            )
        except Exception as exc:
            # Best-effort, but never silent (see session.open).
            logger.warning("session.close KG write failed: %s", exc)