orchestrata_ultimate.session.graph_store

Reference › orchestrata_ultimate › session › graph_store

GraphSessionStore is the cold, archival layer of the dual-storage design: it keeps sessions, facts, portfolios and carryover items in the knowledge graph. The constructor takes a URI, a username and password, and optional driver arguments. The hot layer is RedisSessionStore, SessionSync copies data into this store, and the FactEmitter writes facts to it on a best-effort basis.

The write methods return a boolean or a count instead of raising: persist_session, persist_fact, persist_facts_batch (one transaction), persist_portfolio (merged on its canonical key), link_session_to_portfolio (a BELONGS_TO relationship), link_sessions (a NEXT_SESSION relationship between consecutive sessions), persist_carryover (a work-item node with carryover relationships) and cleanup_session. connected tells whether the graph is reachable and close ends the connection. The read methods are query_session_history, optionally filtered by portfolio and limited to 50 by default, and query_fact_timeline, which returns a session's facts in creation order. The read side for code and spec context is in kg_context.

At a glance

Class diagram

Dependency diagram

API

Classes

GraphSessionStore

Decorators: traces(spec='006', story='US-002', reqs=['REQ-006-FR5-002', 'REQ-006-US-002-004']) · Line 48

Graph-backed session archival storage.

Functions

Source

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

# mypy: ignore-errors
"""Graph session store for cold/archival storage.

Provides persistent session history backed by the operational graph database.
"""

# Spec: 006 - Session Management

import json
import logging
import os
from typing import Any

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.core.graph import BoltSyncRepository, GraphRepositoryConfig
from orchestrata_ultimate.knowledge_graph.canonical_schema import (
    LABEL_PROJECT,
    REL_HAS_SESSION,
)
from orchestrata_ultimate.session.constants import (
    ABOUT,
    BELONGS_TO,
    CARRIED_OVER_FROM,
    CARRIED_OVER_TO,
    FACT_LABEL,
    NEXT_SESSION,
    PORTFOLIO_LABEL,
    SESSION_LABEL,
    WORKITEM_LABEL,
)
from orchestrata_ultimate.session.types import Fact

logger = logging.getLogger(__name__)

# The root Project slug the HAS_SESSION hub edge below MERGEs onto -- same
# default/idiom every other KG writer in this codebase uses
# (gateway/audit.py's own `_project_slug()`, copied verbatim in style). Read
# at call time (RUL-003), never cached at import time.
_DEFAULT_PROJECT_SLUG = "orchestrata-ultimate"


def _project_slug() -> str:
    """Resolve the root ``:Project`` slug the Session hub edge MERGEs onto."""
    return os.environ.get("ORCHESTRATA_PROJECT_ID", _DEFAULT_PROJECT_SLUG)


@traces(spec="006", story="US-002", reqs=["REQ-006-FR5-002", "REQ-006-US-002-004"])
class GraphSessionStore:
    """Graph-backed session archival storage.

    Node schema:
        (:Session {session_id, id, status, goals, started_at, closed_at,
                   portfolio_id, summary})
        (:Fact {id, fact_type, data, source, created_at})
        (:Portfolio {id})
        (:WorkItem {id, text, status, priority})

    Relationship schema:
        (:Session)-[:BELONGS_TO]->(:Portfolio)
        (:Fact)-[:ABOUT]->(:Session)
        (:Session)-[:NEXT_SESSION]->(:Session)
        (:WorkItem)-[:CARRIED_OVER_FROM]->(:Session)
        (:WorkItem)-[:CARRIED_OVER_TO]->(:Session)
    """

    def __init__(
        self,
        uri: str,
        username: str,
        password: str,
        *,
        driver_kwargs: dict[str, Any] | None = None,
    ) -> None:
        self._uri = uri
        self._username = username
        self._password = password
        self._driver_kwargs = dict(driver_kwargs or {})
        self._repo: BoltSyncRepository | None = None
        self._connect()

    @property
    def _driver(self) -> Any | None:
        """Raw-driver read-only escape hatch, kept for external inspection."""
        return self._repo.driver if self._repo is not None else None

    def _connect(self) -> None:
        """Establish graph connection."""
        try:
            repo = BoltSyncRepository(
                GraphRepositoryConfig(
                    uri=self._uri,
                    auth=(self._username, self._password),
                    driver_kwargs=self._driver_kwargs,
                )
            )
            self._repo = repo
            repo.verify_connectivity()
        except Exception as e:
            logger.warning("Graph connection failed: %s", e)
            self._repo = None

    @property
    def connected(self) -> bool:
        """Check if the graph is connected."""
        return self._repo is not None

    @traces(spec="006", story="FR-005", reqs=["REQ-006-FR5-002"])
    def close(self) -> None:
        """Close graph connection."""
        if self._repo:
            self._repo.close()
            self._repo = None

    # ── Session persistence ───────────────────────────────────────

    def persist_session(self, session: dict) -> bool:
        """Write session to the graph.

        Session-254 root-connectivity ruling: this write used to MERGE
        ``:Session`` with no edge at all, leaving it permanently unreachable
        from the root ``:Project`` node. The ``HAS_SESSION`` hub MERGE below
        closes that gap, matching every other Session writer in the repo
        (``gateway/audit.py``, ``kernel/session_manager.py``).
        """
        repo = self._repo
        if not repo:
            return False
        try:
            repo.execute_write(
                f"MERGE (s:{SESSION_LABEL} {{session_id: $id}}) "
                "SET s.id = $id, s.status = $status, s.goals = $goals, "
                "s.started_at = $started_at, s.closed_at = $closed_at, "
                "s.portfolio_id = $portfolio_id, s.summary = $summary "
                "WITH s "
                f"MERGE (p:{LABEL_PROJECT} {{slug: $project}}) "
                f"MERGE (p)-[:{REL_HAS_SESSION}]->(s)",
                {
                    "id": session.get("id", ""),
                    "status": session.get("status", ""),
                    "goals": json.dumps(session.get("goals", [])),
                    "started_at": session.get("started_at", ""),
                    "closed_at": session.get("closed_at", ""),
                    "portfolio_id": session.get("portfolio_id", ""),
                    "summary": session.get("summary", ""),
                    "project": _project_slug(),
                },
            )
            return True
        except Exception as e:
            logger.error("Failed to persist session: %s", e)
            return False

    # ── Fact persistence ──────────────────────────────────────────

    @traces(
        spec="006",
        story="FR-003",
        reqs=["REQ-006-FR3-001", "REQ-006-FR3-002", "REQ-006-FR3-003"],
    )
    def persist_fact(self, session_id: str, fact: Fact) -> bool:
        """Write fact to the graph and link to session."""
        repo = self._repo
        if not repo:
            return False
        try:
            fact_data = fact.to_dict()
            repo.execute_write(
                f"MERGE (f:{FACT_LABEL} {{id: $fact_id}}) "
                "SET f.fact_type = $fact_type, f.data = $data, "
                "f.source = $source, f.created_at = $created_at "
                f"WITH f "
                f"MATCH (s:{SESSION_LABEL} {{session_id: $session_id}}) "
                f"MERGE (f)-[:{ABOUT}]->(s)",
                {
                    "fact_id": fact_data.get("id", ""),
                    "fact_type": fact_data.get("fact_type", ""),
                    "data": json.dumps(fact_data.get("data", {})),
                    "source": fact_data.get("source", ""),
                    "created_at": fact_data.get("created_at", ""),
                    "session_id": session_id,
                },
            )
            return True
        except Exception as e:
            logger.error("Failed to persist fact: %s", e)
            return False

    @traces(spec="006", story="US-004", reqs=["REQ-006-US-004-005"])
    def persist_facts_batch(self, session_id: str, facts: list[Fact]) -> int:
        """Write multiple facts in a single transaction."""
        repo = self._repo
        if not repo:
            return 0
        try:
            prepared = []
            for f in facts:
                fd = f.to_dict()
                prepared.append(
                    {
                        "id": fd.get("id", ""),
                        "fact_type": fd.get("fact_type", ""),
                        "data": json.dumps(fd.get("data", {})),
                        "source": fd.get("source", ""),
                        "created_at": fd.get("created_at", ""),
                    }
                )
            rows = repo.execute_write(
                "UNWIND $facts AS fact_data "
                f"MERGE (f:{FACT_LABEL} {{id: fact_data.id}}) "
                "SET f.fact_type = fact_data.fact_type, f.data = fact_data.data, "
                "f.source = fact_data.source, f.created_at = fact_data.created_at "
                f"WITH f, fact_data "
                f"MATCH (s:{SESSION_LABEL} {{session_id: $session_id}}) "
                f"MERGE (f)-[:{ABOUT}]->(s) "
                "RETURN count(f) AS cnt",
                {"facts": prepared, "session_id": session_id},
            )
            return rows[0]["cnt"] if rows else 0
        except Exception as e:
            logger.error("Failed to persist facts batch: %s", e)
            return 0

    # ── Portfolio persistence ─────────────────────────────────────

    def persist_portfolio(self, portfolio_name: str) -> bool:
        """Create or merge a Portfolio node on its canonical key.

        ``Portfolio``'s canonical merge key is ``name``, not ``id``
        (``canonical_schema.CANONICAL_MERGE_KEYS``, reconciled to the
        authoritative ``specs/008-*/graph-schema.cql`` constraint
        ``portfolio_name_unique``). These three sites keyed on ``id``, so
        every portfolio this store wrote was a canonical-conformance
        violation: ``canonical_schema.check()`` reports it under "Labels
        with nodes lacking their canonical key" (caught live in two worker
        databases, 2026-07-26).
        """
        repo = self._repo
        if not repo:
            return False
        try:
            repo.execute_write(
                f"MERGE (p:{PORTFOLIO_LABEL} {{name: $name}})",
                {"name": portfolio_name},
            )
            return True
        except Exception as e:
            logger.error("Failed to persist portfolio: %s", e)
            return False

    def link_session_to_portfolio(self, session_id: str, portfolio_name: str) -> bool:
        """Create BELONGS_TO relationship."""
        repo = self._repo
        if not repo:
            return False
        try:
            repo.execute_write(
                f"MATCH (s:{SESSION_LABEL} {{session_id: $sid}}) "
                f"MATCH (p:{PORTFOLIO_LABEL} {{name: $pname}}) "
                f"MERGE (s)-[:{BELONGS_TO}]->(p)",
                {"sid": session_id, "pname": portfolio_name},
            )
            return True
        except Exception as e:
            logger.error("Failed to link session to portfolio: %s", e)
            return False

    # ── Session linking ───────────────────────────────────────────

    def link_sessions(self, prev_id: str, next_id: str) -> bool:
        """Create NEXT_SESSION relationship between sessions."""
        repo = self._repo
        if not repo:
            return False
        try:
            repo.execute_write(
                f"MATCH (a:{SESSION_LABEL} {{session_id: $prev}}) "
                f"MATCH (b:{SESSION_LABEL} {{session_id: $next}}) "
                f"MERGE (a)-[:{NEXT_SESSION}]->(b)",
                {"prev": prev_id, "next": next_id},
            )
            return True
        except Exception as e:
            logger.error("Failed to link sessions: %s", e)
            return False

    # ── Query operations ──────────────────────────────────────────

    def query_session_history(
        self, portfolio_name: str | None = None, limit: int = 50
    ) -> list[dict]:
        """Query session history, optionally filtered by portfolio.

        Filters on the canonical ``Portfolio.name`` key — see
        :meth:`persist_portfolio`.
        """
        repo = self._repo
        if not repo:
            return []
        try:
            if portfolio_name:
                rows = repo.execute_read(
                    f"MATCH (s:{SESSION_LABEL})-[:{BELONGS_TO}]->"
                    f"(p:{PORTFOLIO_LABEL} {{name: $pname}}) "
                    "RETURN s ORDER BY s.started_at DESC LIMIT $limit",
                    {"pname": portfolio_name, "limit": limit},
                )
            else:
                rows = repo.execute_read(
                    f"MATCH (s:{SESSION_LABEL}) "
                    "RETURN s ORDER BY s.started_at DESC LIMIT $limit",
                    {"limit": limit},
                )
            return [dict(row["s"]) for row in rows]
        except Exception as e:
            logger.error("Failed to query session history: %s", e)
            return []

    @traces(spec="006", story="US-003", reqs=["REQ-006-US-003-002"])
    def query_fact_timeline(self, session_id: str) -> list[dict]:
        """Get all facts for a session ordered by creation time."""
        repo = self._repo
        if not repo:
            return []
        try:
            rows = repo.execute_read(
                f"MATCH (f:{FACT_LABEL})-[:{ABOUT}]->"
                f"(s:{SESSION_LABEL} {{session_id: $sid}}) "
                "RETURN f ORDER BY f.created_at",
                {"sid": session_id},
            )
            return [dict(row["f"]) for row in rows]
        except Exception as e:
            logger.error("Failed to query fact timeline: %s", e)
            return []

    # ── Carryover operations ──────────────────────────────────────

    @traces(spec="001", story="US-004", reqs=["REQ-001-US-004-028"])
    def persist_carryover(
        self,
        item_id: str,
        text: str,
        status: str,
        priority: str,
        from_session: str,
        to_session: str | None = None,
    ) -> bool:
        """Create a WorkItem node with carryover relationships.

        Both statements run in one transaction (unit_of_work) since the
        pre-seam version ran them inside a single managed transaction
        function.
        """
        repo = self._repo
        if not repo:
            return False
        try:
            with repo.unit_of_work() as uow:
                uow.run(
                    f"MERGE (w:{WORKITEM_LABEL} {{id: $id}}) "
                    "SET w.text = $text, w.status = $status, w.priority = $priority "
                    f"WITH w "
                    f"MATCH (s:{SESSION_LABEL} {{session_id: $from_sid}}) "
                    f"MERGE (w)-[:{CARRIED_OVER_FROM}]->(s)",
                    {
                        "id": item_id,
                        "text": text,
                        "status": status,
                        "priority": priority,
                        "from_sid": from_session,
                    },
                )
                if to_session:
                    uow.run(
                        f"MATCH (w:{WORKITEM_LABEL} {{id: $id}}) "
                        f"MATCH (s:{SESSION_LABEL} {{session_id: $to_sid}}) "
                        f"MERGE (w)-[:{CARRIED_OVER_TO}]->(s)",
                        {"id": item_id, "to_sid": to_session},
                    )
            return True
        except Exception as e:
            logger.error("Failed to persist carryover: %s", e)
            return False

    # ── Cleanup ───────────────────────────────────────────────────

    @traces(spec="006", story="US-002", reqs=["REQ-006-FR5-002", "REQ-006-US-002-004"])
    def cleanup_session(self, session_id: str) -> bool:
        """Remove all data for a session from the graph.

        Both deletes run in one transaction (unit_of_work) since the
        pre-seam version ran them inside a single managed transaction
        function.
        """
        repo = self._repo
        if not repo:
            return False
        try:
            with repo.unit_of_work() as uow:
                uow.run(
                    f"MATCH (f:{FACT_LABEL})-[:{ABOUT}]->"
                    f"(s:{SESSION_LABEL} {{session_id: $sid}}) "
                    "DETACH DELETE f",
                    {"sid": session_id},
                )
                uow.run(
                    f"MATCH (s:{SESSION_LABEL} {{session_id: $sid}}) "
                    "DETACH DELETE s",
                    {"sid": session_id},
                )
            return True
        except Exception as e:
            logger.error("Failed to cleanup session: %s", e)
            return False