orchestrata_ultimate.knowledge_graph.pam_integration

Reference › orchestrata_ultimate › knowledge_graph › pam_integration

PAMIntegration enforces PAM++ ownership rules on knowledge graph nodes (Spec 008, REQ-008-031 and REQ-008-042). It keeps a node-to-agent ownership map and a delegation list in process memory, with an optional prolog_bridge KnowledgePrologBridge as a second authority. set_owner, get_owner, check_ownership and get_owned_nodes manage ownership; delegate, get_delegations and revoke_delegation manage delegated access per operation; clear resets both. enforce_access applies the rules in its docstring: an owner may do anything, read is open to every agent, and a write or delete needs ownership. When the caller is not the owner and the operation is not a read, it asks the Prolog bridge check_can_modify; a Prolog error is logged and access is denied. A violation that the caller escalates is expressed with OwnershipError from exceptions. Nothing is persisted to the graph, and only tests import the module.

At a glance

Class diagram

Dependency diagram

API

Classes

PAMIntegration

Decorators: traces(spec='008', story='US-001', reqs=['REQ-008-031', 'REQ-008-042']) · Line 31

PAM++ ownership enforcement for knowledge graph nodes.

Source

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

# mypy: ignore-errors
"""PAM++ integration for the knowledge graph.

Enforces ownership, access control, and delegation
tracking through the PAM++ governance layer.
"""

from __future__ import annotations

import logging
from typing import TYPE_CHECKING, Any

from orchestrata_ultimate.core.base.traces import traces

if TYPE_CHECKING:
    from orchestrata_ultimate.knowledge_graph.prolog_bridge import (
        KnowledgePrologBridge,
    )

logger = logging.getLogger(__name__)


@traces(
    spec="008",
    story="US-001",
    reqs=[
        "REQ-008-031",
        "REQ-008-042",
    ],
)
class PAMIntegration:
    """PAM++ ownership enforcement for knowledge graph nodes.

    Maps PAM++ agent assignments to graph node ownership,
    enforces access control, and tracks delegation.
    """

    def __init__(self, prolog_bridge: KnowledgePrologBridge | None = None) -> None:
        self._prolog = prolog_bridge
        self._ownership: dict[str, str] = {}  # node_id -> agent_id
        self._delegations: list[dict[str, Any]] = []

    @traces(spec="008", story="US-001", reqs=["REQ-008-042"])
    def set_owner(self, node_id: str, agent_id: str) -> None:
        """Set the owner of a node."""
        self._ownership[node_id] = agent_id
        if self._prolog:
            try:
                self._prolog.assert_ownership(agent_id, node_id)
            except Exception as exc:
                logger.warning("Failed to assert ownership in Prolog: %s", exc)

    def get_owner(self, node_id: str) -> str | None:
        """Get the owner of a node."""
        return self._ownership.get(node_id)

    @traces(spec="001", story="US-003", reqs=["REQ-001-US-003-003"])
    def check_ownership(self, agent_id: str, node_id: str) -> bool:
        """Check if an agent owns a node."""
        owner = self._ownership.get(node_id)
        if owner == agent_id:
            return True
        if self._prolog:
            try:
                return self._prolog.query_ownership(agent_id, node_id)
            except Exception as exc:
                logger.warning(
                    "Prolog ownership query failed for agent %s, node %s: %s",
                    agent_id,
                    node_id,
                    exc,
                    exc_info=True,
                )
        return False

    def get_owned_nodes(self, agent_id: str) -> list[str]:
        """Get all nodes owned by an agent."""
        return [nid for nid, aid in self._ownership.items() if aid == agent_id]

    @traces(
        spec="001", story="US-001", reqs=["REQ-001-US-003-002", "REQ-001-US-003-017"]
    )
    @traces(spec="008", story="US-001", reqs=["REQ-008-031"])
    def enforce_access(self, agent_id: str, node_id: str, operation: str) -> bool:
        """Check if an agent can perform an operation on a node.

        Rules:
        - Owner can do anything
        - Read is allowed for all agents in same scope
        - Write/delete requires ownership
        """
        if self.check_ownership(agent_id, node_id):
            return True
        if operation == "read":
            return True  # Read is open
        if self._prolog:
            try:
                return self._prolog.check_can_modify(agent_id, node_id)
            except Exception as exc:
                logger.warning(
                    "Prolog modify-access check failed for agent %s, node %s: %s",
                    agent_id,
                    node_id,
                    exc,
                    exc_info=True,
                )
        return False

    @traces(spec="001", story="US-003", reqs=["REQ-001-US-003-018"])
    def delegate(
        self, from_agent: str, to_agent: str, node_id: str, operations: list[str]
    ) -> bool:
        """Delegate access to a node."""
        if not self.check_ownership(from_agent, node_id):
            return False
        delegation = {
            "from_agent": from_agent,
            "to_agent": to_agent,
            "node_id": node_id,
            "operations": operations,
        }
        self._delegations.append(delegation)
        return True

    @traces(spec="001", story="US-003", reqs=["REQ-001-US-003-018"])
    def get_delegations(self, agent_id: str) -> list[dict[str, Any]]:
        """Get all delegations for an agent."""
        return [d for d in self._delegations if d["to_agent"] == agent_id]

    @traces(spec="001", story="US-003", reqs=["REQ-001-US-003-018"])
    def revoke_delegation(self, from_agent: str, to_agent: str, node_id: str) -> bool:
        """Revoke a delegation."""
        initial_len = len(self._delegations)
        self._delegations = [
            d
            for d in self._delegations
            if not (
                d["from_agent"] == from_agent
                and d["to_agent"] == to_agent
                and d["node_id"] == node_id
            )
        ]
        return len(self._delegations) < initial_len

    def clear(self) -> None:
        """Clear all ownership and delegation data."""
        self._ownership.clear()
        self._delegations.clear()