orchestrata_ultimate.orchestrator.fact_graph

Reference › orchestrata_ultimate › orchestrator › fact_graph

The graph half of the fact store, split out of fact_store, which is its only importer. It works on an AsyncGraphRepository from core.graph.

build_create_query produces the Cypher text and parameters that create a fact node from its id, predicate, arguments, creation time and optional session id. write_fact executes that write as a best-effort cold path: any exception is caught, logged, counted in a module-level failure counter (read with get_graph_failure_count) and passed to an optional on_failure callback, and nothing is raised to the caller. delete_fact removes one fact node and returns whether it did, and clear_facts removes all fact nodes.

At a glance

Class diagram

No classes; see Functions.

Dependency diagram

API

Functions

Source

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

"""Graph operations for the fact store.

Extracted from fact_store.py to keep RealFactStore under 300 lines.
Provides module-level functions for graph CRUD on Fact nodes.
"""

import json
import logging
import os
from collections.abc import Callable
from typing import Any

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.core.graph import AsyncGraphRepository

logger = logging.getLogger(__name__)

_GRAPH_FAILURE_COUNT = 0

# 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)


def _increment_graph_failures() -> int:
    """Increment and return the graph failure counter.

    Returns:
        Updated failure count.
    """
    global _GRAPH_FAILURE_COUNT
    _GRAPH_FAILURE_COUNT += 1
    return _GRAPH_FAILURE_COUNT


def get_graph_failure_count() -> int:
    """Return the current graph failure count.

    Returns:
        Number of graph failures recorded.
    """
    return _GRAPH_FAILURE_COUNT


@traces(spec="006", story="FR8", reqs=["REQ-006-FR8-001"])
def build_create_query(
    fact_id: str,
    predicate: str,
    arguments: tuple,
    created_at: str,
    session_id: str,
) -> tuple[str, dict[str, Any]]:
    """Build the Cypher query and params for fact creation.

    Args:
        fact_id: Unique fact identifier.
        predicate: Prolog predicate name.
        arguments: Fact arguments.
        created_at: ISO timestamp string.
        session_id: Optional session ID.

    Returns:
        Tuple of (query_string, params_dict).
    """
    args_json = json.dumps(list(arguments))

    # Canonical KG key for Fact is `id` (canonical_schema.NODE_KEYS);
    # `fact_id` is kept as the store's own lookup property — same value.
    if session_id:
        # Session-254 root-connectivity ruling: this write used to MERGE
        # `:Session` with only the inbound `(f)-[:ABOUT]->(s)` edge, leaving
        # the Session itself (and everything else that might hang off 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).
        query = (
            "CREATE (f:Fact {"
            "  id: $fact_id,"
            "  fact_id: $fact_id,"
            "  predicate: $predicate,"
            "  arguments: $arguments,"
            "  created_at: $created_at,"
            "  session_id: $sid"
            "}) "
            "WITH f "
            "MERGE (s:Session {session_id: $sid}) "
            "MERGE (f)-[:ABOUT]->(s) "
            "WITH s "
            "MERGE (p:Project {slug: $project}) "
            "MERGE (p)-[:HAS_SESSION]->(s)"
        )
        params = {
            "fact_id": fact_id,
            "predicate": predicate,
            "arguments": args_json,
            "created_at": created_at,
            "sid": session_id,
            "project": _project_slug(),
        }
    else:
        query = (
            "CREATE (f:Fact {"
            "  id: $fact_id,"
            "  fact_id: $fact_id,"
            "  predicate: $predicate,"
            "  arguments: $arguments,"
            "  created_at: $created_at"
            "})"
        )
        params = {
            "fact_id": fact_id,
            "predicate": predicate,
            "arguments": args_json,
            "created_at": created_at,
        }

    return query, params


@traces(spec="003", story="US-003", reqs=["REQ-003-US-003-001"])
async def write_fact(
    repo: AsyncGraphRepository,
    fact_id: str,
    predicate: str,
    arguments: tuple,
    created_at: str,
    session_id: str,
    on_failure: Callable[[Exception], None] | None,
) -> None:
    """Store a fact in graph (cold path, best effort).

    Args:
        repo: Owned async graph repository.
        fact_id: Unique fact identifier.
        predicate: Prolog predicate name.
        arguments: Fact arguments.
        created_at: ISO timestamp string.
        session_id: Optional session ID to link to.
        on_failure: Optional failure callback.
    """
    # gate-ok: the handler informs the caller through the on_failure(exc) seam
    try:
        query, params = build_create_query(
            fact_id,
            predicate,
            arguments,
            created_at,
            session_id,
        )
        await repo.execute_write(query, params)
    except Exception as exc:
        count = _increment_graph_failures()
        logger.error(
            "Graph cold storage write failed for %s (total failures: %d): %s",
            fact_id,
            count,
            exc,
        )
        if on_failure is not None:
            on_failure(exc)


@traces(spec="006", story="US2", reqs=["REQ-006-US-002-003"])
async def delete_fact(repo: AsyncGraphRepository, fact_id: str) -> bool:
    """Delete a fact from graph (best effort).

    Returns True when the delete landed, False when it failed. "Best effort"
    means the caller is not INTERRUPTED by a graph failure — it never meant
    the caller cannot find out. Returning None left a failed delete
    indistinguishable from a successful one, so contention on the shared
    graph surfaced later as an unexplained read-back instead of at the site
    that failed.

    Args:
        repo: Owned async graph repository.
        fact_id: Fact identifier.
    """
    try:
        await repo.execute_write(
            "MATCH (f:Fact {fact_id: $fact_id}) DELETE f",
            {"fact_id": fact_id},
        )
        return True
    except Exception as exc:
        logger.error(
            "Graph cold storage delete failed for %s: %s",
            fact_id,
            exc,
        )
        return False


async def clear_facts(repo: AsyncGraphRepository) -> bool:
    """Clear all Fact nodes from graph (best effort).

    Returns True when the clear landed, False when it failed — see
    :func:`delete_fact` for why "best effort" still owes the caller an answer.

    Args:
        repo: Owned async graph repository.
    """
    try:
        await repo.execute_write("MATCH (f:Fact) DELETE f")
        return True
    except Exception as exc:
        logger.error("Graph cold storage clear failed: %s", exc)
        return False