feat(agent-mesh): wire GovernedCallable into hypervisor ring enforcement

agent-governance-python/agent-mesh/pyproject.toml

@@ -47,6 +47,7 @@ requires-python = ">=3.11"
 dependencies = [
     # Core dependency redirect
     "agent-governance-toolkit-core>=4.0.0,<5.0",
+    "agent_hypervisor>=4.0.0,<5.0",
     # Original deps kept for backward compat
     "pydantic[email]>=2.5.0,<3.0",
     "cryptography>=46.0.7,<49.0",

agent-governance-python/agent-mesh/src/agentmesh/governance/__init__.py

@@ -7,6 +7,8 @@
 Append-only audit logs with optional external sinks.
 """
 from .govern import govern, GovernedCallable, GovernanceConfig, GovernanceDenied
+from hypervisor.models import ExecutionRing
+from hypervisor.rings.enforcer import ResourceConstraints, ResourceType, RING_CONSTRAINTS, RingCheckResult
 from .approval import (
     ApprovalHandler,
     ApprovalRequest,
@@ -112,6 +114,12 @@
     "GovernedCallable",
     "GovernanceConfig",
     "GovernanceDenied",
+    # Ring enforcement (issue #2667)
+    "ExecutionRing",
+    "ResourceConstraints",
+    "ResourceType",
+    "RING_CONSTRAINTS",
+    "RingCheckResult",
     # Approval workflows (issue #1374)
     "ApprovalHandler",
     "ApprovalRequest",

agent-governance-python/agent-mesh/src/agentmesh/governance/govern.py

@@ -21,13 +21,16 @@
 import os
 import time
 from dataclasses import dataclass
-from typing import Any, Callable, Optional, Union
+from typing import TYPE_CHECKING, Any, Callable, Optional, Union
 
 from .policy import Policy, PolicyDecision, PolicyEngine
 from .audit import AuditLog
 from .approval import ApprovalHandler, ApprovalRequest, AutoRejectApproval
 from .advisory import AdvisoryCheck, AdvisoryDecision
 
+if TYPE_CHECKING:
+    from hypervisor.models import ExecutionRing
+
 logger = logging.getLogger(__name__)
 
 
@@ -42,6 +45,11 @@ class GovernanceConfig:
         audit_file: Path for file-based audit log. None = in-memory only.
         on_deny: Callback when a policy denies an action. Default: raise.
         conflict_strategy: Policy conflict resolution strategy.
+        ring: Optional execution ring for the agent. When set, ring-level
+            resource constraints are enforced before policy evaluation and
+            injected into the evaluation context as ``ring.*`` fields.
+        session_id: Agent session identifier used by RingBreachDetector
+            to track per-session violation rates. Defaults to "".
     """
 
     policy: Union[str, Policy]
@@ -52,6 +60,8 @@ class GovernanceConfig:
     approval_handler: Optional[ApprovalHandler] = None
     advisory: Optional[AdvisoryCheck] = None
     conflict_strategy: str = "deny_overrides"
+    ring: Optional["ExecutionRing"] = None
+    session_id: str = ""
 
 
 class GovernanceDenied(Exception):
@@ -98,13 +108,31 @@ def __init__(self, fn: Callable, config: GovernanceConfig):
         if not loaded.agent and not loaded.agents:
             loaded.agents = ["*"]
 
+        # Ring enforcement — only active when a ring is explicitly configured.
+        self._ring_enforcer: Any = None
+        self._breach_detector: Any = None
+        if config.ring is not None:
+            from hypervisor.rings.enforcer import RingEnforcer
+            from hypervisor.rings.breach_detector import RingBreachDetector
+            self._ring_enforcer = RingEnforcer()
+            self._breach_detector = RingBreachDetector()
+
         functools.update_wrapper(self, fn)
 
     def __call__(self, *args: Any, **kwargs: Any) -> Any:
         """Execute the wrapped function with governance enforcement."""
         # Build evaluation context from kwargs
         context = self._build_context(args, kwargs)
 
+        # Ring enforcement — runs before policy evaluation so a denied ring
+        # never reaches the policy engine.
+        if self._ring_enforcer is not None and self._config.ring is not None:
+            ring_denial = self._check_ring(context)
+            if ring_denial is not None:
+                if self._config.on_deny:
+                    return self._config.on_deny(ring_denial)
+                raise GovernanceDenied(ring_denial)
+
         # Evaluate policy
         start = time.monotonic()
         decision = self._engine.evaluate(self._config.agent_id, context)
@@ -152,6 +180,86 @@ def __call__(self, *args: Any, **kwargs: Any) -> Any:
         # Allowed — execute the wrapped function
         return self._fn(*args, **kwargs)
 
+    def _check_ring(self, context: dict) -> Optional[PolicyDecision]:
+        """Enforce ring-level resource constraints and inject ring context.
+
+        Returns a denial PolicyDecision when the agent's ring forbids the
+        requested resource, or None when access is permitted. Also injects
+        ``ring`` and ``ring_constraints`` into *context* so downstream policy
+        rules can reference them (e.g. ``ring.subprocess_allowed == false``).
+        """
+        from hypervisor.rings.enforcer import RING_CONSTRAINTS, ResourceType
+
+        ring = self._config.ring
+
+        # Circuit-breaker: if this agent/session has tripped the breaker from
+        # prior violations, deny immediately without consulting the ring enforcer.
+        session_id = self._config.session_id or "default"
+        if self._breach_detector.is_breaker_tripped(self._config.agent_id, session_id):
+            return PolicyDecision(
+                allowed=False,
+                action="deny",
+                matched_rule="ring_breaker",
+                reason=(
+                    f"Circuit breaker tripped for agent '{self._config.agent_id}' "
+                    f"— too many ring violations in session '{session_id}'"
+                ),
+            )
+
+        # Infer the resource type from the action in context.
+        action_type = context.get("action", {}).get("type", "").lower()
+        if any(k in action_type for k in ("subprocess", "exec", "shell", "process")):
+            resource_type = ResourceType.SUBPROCESS
+        elif any(k in action_type for k in ("network", "http", "https", "request", "fetch", "url", "web", "api")):
+            resource_type = ResourceType.NETWORK
+        elif any(k in action_type for k in ("file", "write", "filesystem", "path", "disk")):
+            resource_type = ResourceType.FILESYSTEM
+        else:
+            resource_type = ResourceType.TOOL_EXECUTION
+
+        ring_result = self._ring_enforcer.check_resource(ring, resource_type)
+
+        # Always inject ring context so policy rules can reference it.
+        constraints = self._ring_enforcer.get_constraints(ring)
+        context["ring"] = {
+            "level": ring.value,
+            "subprocess_allowed": constraints.subprocess_allowed,
+            "network_allowed": constraints.network_allowed,
+            "filesystem_scope": constraints.filesystem_scope,
+            "filesystem_writable": constraints.filesystem_writable,
+        }
+
+        if ring_result.allowed:
+            return None
+
+        # Find the minimum ring that allows this resource type so the breach
+        # detector receives an accurate ring distance.
+        from hypervisor.models import ExecutionRing
+        called_ring = next(
+            (
+                r for r in (
+                    ExecutionRing.RING_2_STANDARD,
+                    ExecutionRing.RING_1_PRIVILEGED,
+                    ExecutionRing.RING_0_ROOT,
+                )
+                if RING_CONSTRAINTS[r].allows_resource(resource_type)
+            ),
+            ExecutionRing.RING_1_PRIVILEGED,
+        )
+        self._breach_detector.record_call(
+            self._config.agent_id, session_id, ring, called_ring
+        )
+
+        return PolicyDecision(
+            allowed=False,
+            action="deny",
+            matched_rule="ring_enforcement",
+            reason=(
+                f"Ring {ring.value} agent cannot perform "
+                f"{resource_type.value}: {ring_result.reason}"
+            ),
+        )
+
     def _handle_approval(self, decision: PolicyDecision, context: dict) -> PolicyDecision:
         """Route require_approval decisions through the approval handler."""
         handler = self._config.approval_handler or AutoRejectApproval()
@@ -272,6 +380,8 @@ def govern(
     approval_handler: Optional[ApprovalHandler] = None,
     advisory: Optional[AdvisoryCheck] = None,
     conflict_strategy: str = "deny_overrides",
+    ring: Optional["ExecutionRing"] = None,
+    session_id: str = "",
 ) -> GovernedCallable:
     """Wrap any callable with AGT governance — 2-line integration.
 
@@ -307,5 +417,7 @@ def send_email(to, body):
         approval_handler=approval_handler,
         advisory=advisory,
         conflict_strategy=conflict_strategy,
+        ring=ring,
+        session_id=session_id,
     )
     return GovernedCallable(fn, config)

agent-governance-python/agent-mesh/tests/test_govern.py

@@ -216,3 +216,110 @@ def test_govern_multiple_calls(self):
         for i in range(10):
             result = safe(action="read", iteration=i)
             assert result["iteration"] == i
+
+
+# ── Ring enforcement tests ─────────────────────────────────────────
+
+class TestRingEnforcement:
+    """Tests for ring-level resource constraint enforcement in govern()."""
+
+    def test_ring3_denies_subprocess_action(self):
+        """Ring 3 agent cannot invoke a subprocess-type action."""
+        from hypervisor.models import ExecutionRing
+        safe = govern(dummy_tool, policy=ALLOW_ALL_POLICY, ring=ExecutionRing.RING_3_SANDBOX)
+        with pytest.raises(GovernanceDenied) as exc_info:
+            safe(action="subprocess_exec")
+        assert exc_info.value.decision.matched_rule == "ring_enforcement"
+        assert "subprocess" in exc_info.value.decision.reason
+
+    def test_ring3_denies_network_action(self):
+        """Ring 3 agent cannot invoke a network-type action."""
+        from hypervisor.models import ExecutionRing
+        safe = govern(dummy_tool, policy=ALLOW_ALL_POLICY, ring=ExecutionRing.RING_3_SANDBOX)
+        with pytest.raises(GovernanceDenied) as exc_info:
+            safe(action="http_request")
+        assert exc_info.value.decision.matched_rule == "ring_enforcement"
+        assert "network" in exc_info.value.decision.reason
+
+    def test_ring3_allows_tool_execution(self):
+        """Ring 3 agent can invoke generic tool-execution actions."""
+        from hypervisor.models import ExecutionRing
+        safe = govern(dummy_tool, policy=ALLOW_ALL_POLICY, ring=ExecutionRing.RING_3_SANDBOX)
+        result = safe(action="read")
+        assert result["status"] == "executed"
+
+    def test_ring2_allows_subprocess_action(self):
+        """Ring 2 agent is permitted to invoke subprocess-type actions."""
+        from hypervisor.models import ExecutionRing
+        safe = govern(dummy_tool, policy=ALLOW_ALL_POLICY, ring=ExecutionRing.RING_2_STANDARD)
+        result = safe(action="subprocess_exec")
+        assert result["status"] == "executed"
+
+    def test_ring2_allows_network_action(self):
+        """Ring 2 agent is permitted to invoke network-type actions."""
+        from hypervisor.models import ExecutionRing
+        safe = govern(dummy_tool, policy=ALLOW_ALL_POLICY, ring=ExecutionRing.RING_2_STANDARD)
+        result = safe(action="http_request")
+        assert result["status"] == "executed"
+
+    def test_no_ring_no_enforcement(self):
+        """When ring is not set, subprocess actions pass through unchanged."""
+        safe = govern(dummy_tool, policy=ALLOW_ALL_POLICY)
+        result = safe(action="subprocess_exec")
+        assert result["status"] == "executed"
+
+    def test_ring_context_injected_for_policy_rules(self):
+        """ring.* fields are injected into evaluation context when ring is set."""
+        from hypervisor.models import ExecutionRing
+
+        RING_AWARE_POLICY = """
+apiVersion: governance.toolkit/v1
+name: ring-aware
+default_action: deny
+rules:
+  - name: allow-read-only-ring
+    condition: "action.type == 'read'"
+    action: allow
+"""
+        safe = govern(dummy_tool, policy=RING_AWARE_POLICY, ring=ExecutionRing.RING_3_SANDBOX)
+        # Policy allows reads; ring 3 allows tool_execution — should pass
+        result = safe(action="read")
+        assert result["status"] == "executed"
+
+    def test_ring3_on_deny_callback_called(self):
+        """on_deny callback receives the ring denial decision."""
+        from hypervisor.models import ExecutionRing
+        denied = []
+        safe = govern(
+            dummy_tool,
+            policy=ALLOW_ALL_POLICY,
+            ring=ExecutionRing.RING_3_SANDBOX,
+            on_deny=lambda d: denied.append(d) or {"status": "denied"},
+        )
+        result = safe(action="subprocess_exec")
+        assert result["status"] == "denied"
+        assert len(denied) == 1
+        assert denied[0].matched_rule == "ring_enforcement"
+
+    def test_ring_denial_does_not_reach_policy_engine(self):
+        """A ring-denied action never reaches policy evaluation."""
+        from hypervisor.models import ExecutionRing
+
+        # Policy would allow everything — ring should deny first
+        safe = govern(dummy_tool, policy=ALLOW_ALL_POLICY, ring=ExecutionRing.RING_3_SANDBOX)
+        with pytest.raises(GovernanceDenied) as exc_info:
+            safe(action="subprocess_exec")
+        # matched_rule comes from ring layer, not from any policy rule
+        assert exc_info.value.decision.matched_rule == "ring_enforcement"
+
+    def test_circuit_breaker_trips_after_repeated_violations(self):
+        """Repeated ring violations trip the circuit breaker."""
+        from hypervisor.rings.breach_detector import RingBreachDetector
+        from hypervisor.models import ExecutionRing
+
+        # Use a detector with a very low baseline so the breaker trips quickly
+        detector = RingBreachDetector(baseline_rate=0.01)
+        # Simulate rapid calls from ring 3 attempting ring 1 access
+        for _ in range(50):
+            detector.record_call("agent-x", "sess-1", ExecutionRing.RING_3_SANDBOX, ExecutionRing.RING_1_PRIVILEGED)
+        assert detector.is_breaker_tripped("agent-x", "sess-1")