orchestrata_ultimate.knowledge_graph.prolog_bridge

Reference › orchestrata_ultimate › knowledge_graph › prolog_bridge

KnowledgePrologBridge adapts the core Prolog bridge to knowledge graph governance questions: who owns a node and who may modify it (Spec 008 REQ-008-035, -044, -045 and Spec 015 REQ-015-150). It wraps an optional PrologBridge and reports available according to whether one was supplied. assert_ownership asserts an owns(agent, node) fact with session scope and returns the fact id. query_ownership, check_can_modify (which queries the Prolog predicate knowledge:can_modify/2), query_transitive_deps and check_in_scope ask the Prolog engine the corresponding questions. Every method degrades safely: with no bridge, or when Prolog raises, it logs a warning and returns None, False or an empty list, so the caller is denied rather than crashing. The only importer is pam_integration. It does not query the graph database.

At a glance

Class diagram

Dependency diagram

API

Classes

KnowledgePrologBridge

Decorators: traces(spec='008', story='US-001', reqs=['REQ-008-035', 'REQ-008-044', 'REQ-008-045']), traces(spec='015', story='US-001', reqs=['REQ-015-150']) · Line 26

Prolog bridge for knowledge graph governance.

Source

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

# mypy: ignore-errors
"""Knowledge-specific Prolog bridge.

Wraps PrologBridge for knowledge graph operations:
ownership, dependency, scope, and modification checks.
"""

# Spec: 008 - Knowledge Graph

from __future__ import annotations

import logging
from typing import TYPE_CHECKING

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.core.enums import FactScope

if TYPE_CHECKING:
    from orchestrata_ultimate.infra.prolog_bridge import PrologBridge

logger = logging.getLogger(__name__)


@traces(spec="008", story="US-001", reqs=["REQ-008-035", "REQ-008-044", "REQ-008-045"])
@traces(spec="015", story="US-001", reqs=["REQ-015-150"])
class KnowledgePrologBridge:
    """Prolog bridge for knowledge graph governance.

    Provides ownership, dependency, scope, and access
    control predicates via the Prolog engine.
    """

    @traces(spec="015", story="US-008-001", reqs=["REQ-015-152"])
    def __init__(self, prolog_bridge: PrologBridge | None = None) -> None:
        self._bridge = prolog_bridge

    @property
    @traces(spec="008", story="US-001", reqs=["REQ-008-045"])
    def available(self) -> bool:
        return self._bridge is not None

    @traces(spec="008", story="US-001", reqs=["REQ-008-044"])
    @traces(spec="015", story="US-001", reqs=["REQ-015-153"])
    def assert_ownership(self, agent_id: str, node_id: str) -> str | None:
        """Assert that an agent owns a node."""
        if not self._bridge:
            return None
        try:
            return self._bridge.assert_fact(
                "owns", [agent_id, node_id], scope=FactScope.SESSION
            )
        except Exception as exc:
            logger.warning("Failed to assert ownership: %s", exc)
            return None

    @traces(spec="008", story="US-001", reqs=["REQ-008-044"])
    @traces(spec="015", story="US-001", reqs=["REQ-015-154"])
    def query_ownership(self, agent_id: str, node_id: str) -> bool:
        """Query if an agent owns a node."""
        if not self._bridge:
            return False
        try:
            results = list(
                self._bridge._prolog.query(f"knowledge:owns('{agent_id}', '{node_id}')")
            )
            return len(results) > 0
        except Exception as exc:
            logger.warning(
                "Prolog owns/2 query failed for agent %s, node %s: %s",
                agent_id,
                node_id,
                exc,
                exc_info=True,
            )
            return False

    @traces(spec="008", story="US-001", reqs=["REQ-008-016", "REQ-008-035"])
    @traces(spec="015", story="US-001", reqs=["REQ-015-155"])
    def check_can_modify(self, agent_id: str, node_id: str) -> bool:
        """Check if an agent can modify a node."""
        if not self._bridge:
            return False
        try:
            results = list(
                self._bridge._prolog.query(
                    f"knowledge:can_modify('{agent_id}', '{node_id}')"
                )
            )
            return len(results) > 0
        except Exception as exc:
            logger.warning(
                "Prolog can_modify/2 query failed for agent %s, node %s: %s",
                agent_id,
                node_id,
                exc,
                exc_info=True,
            )
            return False

    @traces(spec="008", story="US-001", reqs=["REQ-008-045"])
    @traces(spec="015", story="US-001", reqs=["REQ-015-157"])
    def query_transitive_deps(self, node_id: str) -> list[str]:
        """Get transitive dependencies via Prolog."""
        if not self._bridge:
            return []
        try:
            results = list(
                self._bridge._prolog.query(
                    f"knowledge:depends_on_transitive('{node_id}', Dep)"
                )
            )
            return [str(r["Dep"]) for r in results]
        except Exception as exc:
            logger.warning(
                "Prolog depends_on_transitive/2 query failed for node %s: %s",
                node_id,
                exc,
                exc_info=True,
            )
            return []

    @traces(spec="008", story="US-001", reqs=["REQ-008-033"])
    @traces(spec="015", story="US-001", reqs=["REQ-015-158"])
    def check_in_scope(self, agent_id: str, node_id: str) -> bool:
        """Check if a node is within an agent scope."""
        if not self._bridge:
            return True  # Default allow if no Prolog
        try:
            results = list(
                self._bridge._prolog.query(
                    f"knowledge:in_scope('{agent_id}', '{node_id}')"
                )
            )
            return len(results) > 0
        except Exception as exc:
            logger.warning(
                "Prolog in_scope/2 query failed for agent %s, node %s: %s",
                agent_id,
                node_id,
                exc,
                exc_info=True,
            )
            return True  # Default allow on error