orchestrata_ultimate.knowledge_graph.interfaces

Reference › orchestrata_ultimate › knowledge_graph › interfaces

Runtime-checkable Protocol contracts for the knowledge graph package, so that components depend on shapes rather than concrete classes. IKnowledgeGraph covers node CRUD, relationship creation and raw Cypher query (REQ-008-001), and is the shape graph_client implements. ICodeAnalyzer has analyze_file and analyze_project (REQ-008-098, -099), matching ast_analyzer. ITemporalManager has get_state_at and get_history (REQ-008-002, -029), matching temporal. IQueryEngine has query, multi_hop and search (REQ-008-004, -012), matching query_engine. SynchronizerProtocol is a package-local port describing the part of FileSynchronizer that anomaly_detector needs: its parser, config, file-hash map and sync_file. The module holds no logic.

At a glance

Class diagram

Dependency diagram

API

Classes

SynchronizerProtocol (protocol)

Bases: Protocol · Decorators: runtime_checkable · Line 24

Package-local port for the FileSynchronizer shape AnomalyDetector needs.

IKnowledgeGraph (protocol)

Bases: Protocol · Decorators: runtime_checkable, traces(spec='008', story='US-001', reqs=['REQ-008-001']) · Line 48

Protocol for knowledge graph CRUD operations.

ICodeAnalyzer (protocol)

Bases: Protocol · Decorators: runtime_checkable, traces(spec='008', story='US-001', reqs=['REQ-008-098', 'REQ-008-099']) · Line 97

Protocol for code analysis.

ITemporalManager (protocol)

Bases: Protocol · Decorators: runtime_checkable, traces(spec='008', story='US-001', reqs=['REQ-008-002', 'REQ-008-029']) · Line 111

Protocol for temporal versioning.

IQueryEngine (protocol)

Bases: Protocol · Decorators: runtime_checkable, traces(spec='008', story='US-001', reqs=['REQ-008-004', 'REQ-008-012']) · Line 126

Protocol for query operations.

Source

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

"""Knowledge Graph protocol interfaces.

Defines the abstract contracts for knowledge graph operations.
"""

# Spec: 008 - Knowledge Graph

from typing import Any, Protocol, runtime_checkable

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.knowledge_graph.code_parser import CodeParser
from orchestrata_ultimate.knowledge_graph.config import SyncConfig
from orchestrata_ultimate.knowledge_graph.types import (
    ChangeEntry,
    ContextBundle,
    FileAnalysis,
    Node,
    ProjectAnalysis,
    SyncResult,
)


@runtime_checkable
class SynchronizerProtocol(Protocol):
    """Package-local port for the FileSynchronizer shape AnomalyDetector needs.

    Extracted to break the anomaly_detector <-> synchronizer import cycle:
    AnomalyDetector depends on this structural shape instead of importing the
    concrete ``FileSynchronizer`` class directly.
    """

    _parser: CodeParser
    _config: SyncConfig
    _file_hashes: dict[str, str]

    @traces(spec="008", story="US-001", reqs=["REQ-008-021", "REQ-008-023"])
    def sync_file(self, file_path: str) -> SyncResult:
        """Sync a single file to the knowledge graph."""
        ...


@runtime_checkable
@traces(
    spec="008",
    story="US-001",
    reqs=["REQ-008-001"],
)
class IKnowledgeGraph(Protocol):
    """Protocol for knowledge graph CRUD operations."""

    async def create_node(self, node: Node) -> str:
        """Create a node. Returns node ID."""
        ...

    async def get_node(self, node_id: str, node_type: str | None = None) -> Node | None:
        """Get a node by ID.

        node_type: optional label hint for a labeled index seek instead of
        an unlabeled AllNodesScan -- see graph_client.GraphClient.get_node.
        """
        ...

    async def update_node(
        self,
        node_id: str,
        properties: dict[str, Any],
        node_type: str | None = None,
    ) -> bool:
        """Update node properties."""
        ...

    async def delete_node(self, node_id: str, node_type: str | None = None) -> bool:
        """Delete a node."""
        ...

    async def create_relationship(
        self,
        source_id: str,
        target_id: str,
        rel_type: str,
        properties: dict[str, Any] | None = None,
        source_type: str | None = None,
        target_type: str | None = None,
    ) -> bool:
        """Create a relationship between two nodes."""
        ...

    async def query(
        self, cypher: str, parameters: dict[str, Any] | None = None
    ) -> list[dict[str, Any]]:
        """Execute a Cypher query."""
        ...


@runtime_checkable
@traces(spec="008", story="US-001", reqs=["REQ-008-098", "REQ-008-099"])
class ICodeAnalyzer(Protocol):
    """Protocol for code analysis."""

    def analyze_file(self, file_path: str) -> FileAnalysis:
        """Analyze a single file."""
        ...

    def analyze_project(self, project_path: str) -> ProjectAnalysis:
        """Analyze an entire project."""
        ...


@runtime_checkable
@traces(spec="008", story="US-001", reqs=["REQ-008-002", "REQ-008-029"])
class ITemporalManager(Protocol):
    """Protocol for temporal versioning."""

    @traces(spec="008", story="US-001", reqs=["REQ-008-028"])
    async def get_state_at(self, node_id: str, timestamp: str) -> Node | None:
        """Get the state of a node at a specific time."""
        ...

    async def get_history(self, node_id: str, limit: int = 100) -> list[ChangeEntry]:
        """Get the change history for a node."""
        ...


@runtime_checkable
@traces(spec="008", story="US-001", reqs=["REQ-008-004", "REQ-008-012"])
class IQueryEngine(Protocol):
    """Protocol for query operations."""

    async def query(
        self, cypher: str, parameters: dict[str, Any] | None = None
    ) -> list[dict[str, Any]]:
        """Execute a query."""
        ...

    @traces(spec="008", story="US-001", reqs=["REQ-008-005"])
    async def multi_hop(self, start_id: str, depth: int = 3) -> ContextBundle:
        """Multi-hop traversal from a starting node."""
        ...

    async def search(
        self, query: str, node_type: str | None = None, limit: int = 100
    ) -> list[Node]:
        """Search for nodes by text."""
        ...