orchestrata_ultimate.gateway.handlers.knowledge

Reference › orchestrata_ultimate › gateway › handlers › knowledge

KnowledgeHandler gives clients read access to the operational knowledge graph through the knowledge.* RPCs. rpc creates it, and server attaches the graph repository (set_graph_repository) and a query engine (set_query_engine). Results are converted to JSON with the helper in utils. See gateway.

• handle_query runs a Cypher query supplied by the client. It requires a query (code -32602 otherwise) and rejects any text matching the mutation pattern (create, merge, delete, set, remove), so the call is read-only.
• handle_nodes lists nodes of a given node_type and handle_relationships lists relationships of a rel_type. Both check the name against a safe-identifier pattern and cap the limit at 1000.
• handle_rtm_progress counts verified requirements against the total and returns a percentage. handle_coverage_history returns recent coverage values from test runs for a sparkline. handle_volk_work_context and handle_component_work_context return a volk's or a component's related work through the query engine (SPEC-008).

Every query runs under a timeout (10 seconds by default). A missing graph yields an empty result with an error text, and a timeout or failure raises RPCError.

At a glance

Class diagram

Dependency diagram

API

Classes

KnowledgeHandler

Line 22

Knowledge graph read-only query methods.

Source

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

"""Knowledge RPC handlers -- query, nodes, relationships.

Implements SPEC-014 FR-019 knowledge graph methods.
"""

import asyncio
import re
from typing import Any

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.gateway.exceptions import RPCError
from orchestrata_ultimate.gateway.handlers.utils import _neo4j_value_to_dict

_MUTATION_PATTERN = re.compile(
    r"\b(MERGE|CREATE|DELETE|SET|REMOVE|DROP|DETACH)\b",
    re.IGNORECASE,
)

_SAFE_IDENTIFIER = re.compile(r"^[A-Za-z_][A-Za-z0-9_]*$")


class KnowledgeHandler:
    """Knowledge graph read-only query methods."""

    def __init__(
        self, graph_repository: Any = None, query_timeout: float = 10.0
    ) -> None:
        """Wire handler to knowledge subsystems.

        Args:
            graph_repository: An AsyncGraphRepository instance.
            query_timeout: Per-query budget in seconds for every
                ``asyncio.wait_for`` in this handler. The seam exists so
                the black-hole-driver timeout tests can prove the real
                TimeoutError path in ~1s instead of the production 10s
                (five such tests cost 50s of suite time before this).
        """
        self._graph_repository = graph_repository
        self._query_timeout = query_timeout
        self._query_engine: Any = None

    def set_graph_repository(self, graph_repository: Any) -> None:
        """Inject the operational-graph repository for knowledge queries.

        Args:
            graph_repository: An AsyncGraphRepository instance.
        """
        self._graph_repository = graph_repository

    def set_query_engine(self, query_engine: Any) -> None:
        """Inject the QueryEngine for volk/component work-context RPCs.

        Args:
            query_engine: A QueryEngine instance (constructed over the same
                Neo4j driver as the graph repository).
        """
        self._query_engine = query_engine

    @traces(spec="016", story="US-009", reqs=["REQ-016-205"])
    @traces(
        spec="014",
        story="US-010",
        reqs=["REQ-014-151", "REQ-014-153", "REQ-014-154"],
    )
    async def handle_query(
        self,
        connection_id: str,
        params: dict[str, Any],
    ) -> dict[str, Any]:
        """Handle knowledge.query -- execute read-only Cypher.

        Args:
            connection_id: Caller connection ID.
            params: query (Cypher string).

        Returns:
            Query results as list of records.
        """
        query = params.get("query", "")
        if not query:
            raise RPCError(
                "Missing 'query' parameter",
                rpc_code=-32602,
            )
        if _MUTATION_PATTERN.search(query):
            raise RPCError(
                "Mutation queries are not allowed (no MERGE/CREATE/DELETE/SET/REMOVE)"
            )
        if not self._graph_repository:
            return {
                "records": [],
                "error": "Graph unavailable",
            }
        try:
            results = await asyncio.wait_for(
                self._execute_cypher(query),
                timeout=self._query_timeout,
            )
            return {"records": results}
        except TimeoutError:
            raise RPCError(
                f"Knowledge query timed out ({self._query_timeout:g}s limit)"
            )
        except Exception as exc:
            raise RPCError(f"Knowledge query failed: {exc}")

    @traces(spec="016", story="US-001", reqs=["REQ-016-006"])
    @traces(
        spec="014",
        story="US-010",
        reqs=["REQ-014-156"],
    )
    async def handle_nodes(
        self,
        connection_id: str,
        params: dict[str, Any],
    ) -> dict[str, Any]:
        """Handle knowledge.nodes -- list nodes by type.

        Args:
            connection_id: Caller connection ID.
            params: node_type, offset, limit.

        Returns:
            List of nodes.
        """
        node_type = params.get("node_type") or params.get("nodeIds", "")
        if not node_type:
            raise RPCError(
                "Missing 'node_type' parameter",
                rpc_code=-32602,
            )
        if not _SAFE_IDENTIFIER.match(node_type):
            raise RPCError("Invalid node_type")
        limit = min(int(params.get("limit", 50)), 1000)
        offset = max(int(params.get("offset", 0)), 0)
        query = f"MATCH (n:{node_type}) RETURN properties(n) as node ORDER BY n.name ASC SKIP {offset} LIMIT {limit}"
        if not self._graph_repository:
            return {
                "nodes": [],
                "error": "Graph unavailable",
            }
        try:
            results = await asyncio.wait_for(
                self._execute_cypher(query),
                timeout=self._query_timeout,
            )
            # Unwrap "node" alias to return flat node objects
            nodes = [row.get("node", row) for row in results]
            return {"nodes": nodes}
        except TimeoutError:
            raise RPCError(
                f"Knowledge query timed out ({self._query_timeout:g}s limit)"
            )
        except Exception as exc:
            raise RPCError(f"Knowledge query failed: {exc}")

    @traces(
        spec="014",
        story="US-010",
        reqs=["REQ-014-157"],
    )
    async def handle_relationships(
        self,
        connection_id: str,
        params: dict[str, Any],
    ) -> dict[str, Any]:
        """Handle knowledge.relationships -- list by type.

        Args:
            connection_id: Caller connection ID.
            params: rel_type.

        Returns:
            List of relationships.
        """
        rel_type = params.get("rel_type", "")
        if not rel_type:
            raise RPCError(
                "Missing 'rel_type' parameter",
                rpc_code=-32602,
            )
        if not _SAFE_IDENTIFIER.match(rel_type):
            raise RPCError("Invalid rel_type")
        limit = min(int(params.get("limit", 50)), 1000)
        query = f"MATCH ()-[r:{rel_type}]->() RETURN r LIMIT {limit}"
        if not self._graph_repository:
            return {
                "relationships": [],
                "error": "Graph unavailable",
            }
        try:
            results = await asyncio.wait_for(
                self._execute_cypher(query),
                timeout=self._query_timeout,
            )
            # Unwrap "r" alias to return flat relationship objects
            relationships = [row.get("r", row) for row in results]
            return {"relationships": relationships}
        except TimeoutError:
            raise RPCError(
                f"Knowledge query timed out ({self._query_timeout:g}s limit)"
            )
        except Exception as exc:
            raise RPCError(f"Knowledge query failed: {exc}")

    @traces(spec="008", story="US-001", reqs=["REQ-008-005"])
    @traces(spec="014", story="US-010", reqs=["REQ-014-151"])
    async def handle_volk_work_context(
        self,
        connection_id: str,
        params: dict[str, Any],
    ) -> dict[str, Any]:
        """Handle knowledge.volk_work_context -- a volk's owned components,
        decisions, and generated artifacts (SPEC-008).

        Args:
            connection_id: Caller connection ID.
            params: volk (volk alias).

        Returns:
            The volk's components, decisions, and artifacts.
        """
        volk = params.get("volk", "")
        if not volk:
            raise RPCError(
                "Missing 'volk' parameter",
                rpc_code=-32602,
            )
        if self._query_engine is None:
            return {
                "volk": volk,
                "components": [],
                "decisions": [],
                "artifacts": [],
                "error": "Graph unavailable",
            }
        try:
            await asyncio.wait_for(
                self._query_engine.ensure_connected(),
                timeout=self._query_timeout,
            )
            result: dict[str, Any] = await asyncio.wait_for(
                self._query_engine.volk_work_context(volk),
                timeout=self._query_timeout,
            )
            return result
        except TimeoutError:
            raise RPCError(
                f"Knowledge query timed out ({self._query_timeout:g}s limit)"
            )
        except Exception as exc:
            raise RPCError(f"Knowledge query failed: {exc}")

    @traces(spec="008", story="US-003", reqs=["REQ-008-012"])
    @traces(spec="014", story="US-010", reqs=["REQ-014-151"])
    async def handle_component_work_context(
        self,
        connection_id: str,
        params: dict[str, Any],
    ) -> dict[str, Any]:
        """Handle knowledge.component_work_context -- everything a volk needs
        to resume work on a component it owns (SPEC-008 flagship query).

        Args:
            connection_id: Caller connection ID.
            params: volk (volk alias), component (bare component name).

        Returns:
            The component, applicable decisions, in-progress tasks, and
            their sessions.
        """
        volk = params.get("volk", "")
        if not volk:
            raise RPCError(
                "Missing 'volk' parameter",
                rpc_code=-32602,
            )
        component = params.get("component", "")
        if not component:
            raise RPCError(
                "Missing 'component' parameter",
                rpc_code=-32602,
            )
        if self._query_engine is None:
            return {
                "volk": volk,
                "component_name": component,
                "component": None,
                "decisions": [],
                "tasks": [],
                "sessions": [],
                "error": "Graph unavailable",
            }
        try:
            await asyncio.wait_for(
                self._query_engine.ensure_connected(),
                timeout=self._query_timeout,
            )
            result: dict[str, Any] = await asyncio.wait_for(
                self._query_engine.component_work_context(volk, component),
                timeout=self._query_timeout,
            )
            return result
        except TimeoutError:
            raise RPCError(
                f"Knowledge query timed out ({self._query_timeout:g}s limit)"
            )
        except Exception as exc:
            raise RPCError(f"Knowledge query failed: {exc}")

    @traces(spec="016", story="US-007", reqs=["REQ-016-153"])
    async def handle_coverage_history(
        self,
        connection_id: str,
        params: dict[str, Any],
    ) -> dict[str, Any]:
        """Return coverage history for sparkline.

        Args:
            connection_id: Caller connection ID.
            params: Optional limit.

        Returns:
            History list of date/value dicts.
        """
        if not self._graph_repository:
            return {"history": []}
        limit = min(int(params.get("limit", 20)), 100)
        query = f"""
        MATCH (t:TestRun)
        WHERE t.coverage IS NOT NULL
        RETURN t.date AS date, t.coverage AS value
        ORDER BY t.date DESC
        LIMIT {limit}
        """
        try:
            results = await asyncio.wait_for(
                self._execute_cypher(query),
                timeout=self._query_timeout,
            )
            history = [
                {"date": row.get("date", ""), "value": row.get("value", 0.0)}
                for row in results
            ]
            return {"history": history}
        except TimeoutError:
            raise RPCError(
                f"Coverage history query timed out ({self._query_timeout:g}s limit)"
            )
        except Exception as exc:
            raise RPCError(f"Coverage history query failed: {exc}")

    @traces(spec="016", story="US-007", reqs=["REQ-016-162"])
    async def handle_rtm_progress(
        self,
        connection_id: str,
        params: dict[str, Any],
    ) -> dict[str, Any]:
        """Return requirements traceability matrix progress.

        Args:
            connection_id: Caller connection ID.
            params: Ignored.

        Returns:
            Verified count, total count, and percentage.
        """
        if not self._graph_repository:
            return {"verified": 0, "total": 0, "percentage": 0.0}
        try:
            total_results = await asyncio.wait_for(
                self._execute_cypher("MATCH (r:Requirement) RETURN count(r) AS total"),
                timeout=self._query_timeout,
            )
            total = total_results[0].get("total", 0) if total_results else 0

            verified_results = await asyncio.wait_for(
                self._execute_cypher(
                    "MATCH (r:Requirement {status: 'verified'}) RETURN count(r) AS verified"
                ),
                timeout=self._query_timeout,
            )
            verified = verified_results[0].get("verified", 0) if verified_results else 0

            percentage = round((verified / total * 100), 1) if total > 0 else 0.0
            return {"verified": verified, "total": total, "percentage": percentage}
        except TimeoutError:
            raise RPCError(
                f"RTM progress query timed out ({self._query_timeout:g}s limit)"
            )
        except Exception as exc:
            raise RPCError(f"RTM progress query failed: {exc}")

    @traces(spec="014", story="US-010", reqs=["REQ-014-155"])
    async def _execute_cypher(
        self,
        query: str,
    ) -> list[dict[str, Any]]:
        """Execute a read-only Cypher query against the graph.

        Args:
            query: Validated read-only Cypher string.

        Returns:
            List of record dicts from the query result.
        """
        rows = await self._graph_repository.execute_read(query)
        return [
            {key: _neo4j_value_to_dict(val) for key, val in row.items()} for row in rows
        ]