feat(governance): enforcement-mode config, policy models, deps

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

Author: aditik0303

pyproject.toml

@@ -5,7 +5,13 @@ description = "Runtime abstractions and interfaces for building agents and autom
 readme = { file = "README.md", content-type = "text/markdown" }
 requires-python = ">=3.11"
 dependencies = [
-  "uipath-core>=0.5.17, <0.6.0"
+  "uipath-core>=0.5.18, <0.6.0",
+  # Governance native-evaluator deps. Live here because the native
+  # evaluator implementation lives in uipath.runtime.governance.native;
+  # uipath-core only carries the small governance contracts.
+  "pyyaml>=6.0",
+  "vaderSentiment>=3.3.2",   # sentiment_concern (A.3.3)
+  "chardet>=5.2.0",          # encoding_concern (A.7.4)
 ]
 classifiers = [
   "Intended Audience :: Developers",
@@ -40,6 +46,7 @@ dev = [
   "pytest-cov>=4.1.0",
   "pytest-mock>=3.11.1",
   "pre-commit>=4.1.0",
+  "types-PyYAML>=6.0",
 ]
 
 [tool.hatch.build.targets.wheel]
@@ -83,6 +90,25 @@ no_implicit_reexport = true
 
 disallow_untyped_defs = false
 
+# Third-party governance-evaluator libs have no type stubs / py.typed marker
+[[tool.mypy.overrides]]
+module = [
+  "yaml",
+  "vaderSentiment.*",
+  "chardet",
+  "price_parser",
+  # uipath.platform.common is imported lazily from traces.py / audit
+  # sinks to read UiPathConfig context attributes. It's first-party but
+  # not a uipath-runtime dep, so its stubs aren't installable here.
+  "uipath.platform.*",
+  # Optional framework adapters; the absence of the framework simply
+  # means the adapter no-ops at import time.
+  "agents",
+  "langchain_core.*",
+  "langgraph.*",
+]
+ignore_missing_imports = true
+
 [tool.pydantic-mypy]
 init_forbid_extra = true
 init_typed = true

src/uipath/runtime/governance/config.py

@@ -0,0 +1,78 @@
+"""Runtime-level governance enforcement-mode state.
+
+The feature-flag gate (``is_governance_enabled``) lives in
+:mod:`uipath.core.governance.config` because it is process-level and
+must be resolvable by callers that do not depend on
+``uipath-runtime``. The enforcement mode is *per-policy* — set by the
+backend on each policy fetch via the ``/runtime/policy`` endpoint —
+and therefore lives here in the runtime package alongside the policy
+loader that applies it.
+"""
+
+from __future__ import annotations
+
+import logging
+import os
+from enum import Enum
+
+logger = logging.getLogger(__name__)
+
+ENV_ENFORCEMENT_MODE = "UIPATH_GOVERNANCE_MODE"
+
+
+class EnforcementMode(str, Enum):
+    """Governance enforcement modes."""
+
+    AUDIT = "audit"  # Evaluate and log; never block.
+    ENFORCE = "enforce"  # Block on DENY rules.
+    DISABLED = "disabled"  # Skip evaluation entirely.
+
+
+_enforcement_mode: EnforcementMode | None = None
+
+
+def get_enforcement_mode() -> EnforcementMode:
+    """Return the current enforcement mode.
+
+    The mode is cached after first read. Resolution order:
+
+    1. A value previously set via :func:`set_enforcement_mode` (the
+       policy loader calls this with the backend-supplied mode on every
+       successful policy fetch — that's the canonical source).
+    2. ``UIPATH_GOVERNANCE_MODE`` env var (developer override).
+    3. Default :attr:`EnforcementMode.AUDIT` — evaluate and log without
+       blocking. The wrapper attaches at runtime construction so the
+       background policy fetch can run; if the backend returns
+       ``disabled``, ``set_enforcement_mode`` flips the cache and
+       subsequent ``evaluate()`` calls short-circuit at evaluator.py:332.
+       Defaulting to AUDIT avoids the chicken-and-egg where a DISABLED
+       default would short-circuit before the policy fetch could ever
+       opt the tenant in.
+    """
+    global _enforcement_mode
+    if _enforcement_mode is not None:
+        return _enforcement_mode
+
+    mode_str = os.getenv(ENV_ENFORCEMENT_MODE, "audit").lower()
+    try:
+        _enforcement_mode = EnforcementMode(mode_str)
+    except ValueError:
+        _enforcement_mode = EnforcementMode.AUDIT
+
+    return _enforcement_mode
+
+
+def set_enforcement_mode(mode: EnforcementMode) -> None:
+    """Set the enforcement mode programmatically.
+
+    The policy loader calls this with the backend-supplied mode on each
+    fetch so the evaluator picks up the platform-controlled value.
+    """
+    global _enforcement_mode
+    _enforcement_mode = mode
+
+
+def reset_enforcement_mode() -> None:
+    """Clear cached enforcement mode (intended for tests)."""
+    global _enforcement_mode
+    _enforcement_mode = None

src/uipath/runtime/governance/native/models.py

@@ -0,0 +1,153 @@
+"""Native policy model.
+
+Rules, checks, conditions and pack indexes consumed by
+:class:`uipath.runtime.governance.native.evaluator.GovernanceEvaluator`.
+
+These are the inputs of the native evaluator. The evaluator-agnostic
+*output* types (``Action``, ``AuditRecord``, …) live in
+:mod:`uipath.core.governance.models`.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from enum import Enum
+from typing import Any
+
+from uipath.core.governance.models import Action, LifecycleHook
+
+
+class Severity(Enum):
+    """Rule severity levels."""
+
+    LOW = "low"
+    MEDIUM = "medium"
+    HIGH = "high"
+    CRITICAL = "critical"
+
+
+@dataclass
+class Condition:
+    """A single condition within a rule check."""
+
+    operator: str
+    field: str
+    value: Any
+    negate: bool = False
+
+
+@dataclass
+class Check:
+    """A check within a rule - contains conditions and action."""
+
+    conditions: list[Condition]
+    action: Action = Action.DENY
+    message: str = ""
+    logic: str = "all"  # "all" (AND) or "any" (OR)
+
+
+@dataclass
+class Rule:
+    """A compliance rule with checks evaluated at a specific lifecycle hook."""
+
+    rule_id: str
+    name: str
+    clause: str
+    hook: LifecycleHook
+    action: Action
+    severity: Severity = Severity.HIGH
+    checks: list[Check] = field(default_factory=list)
+    enabled: bool = True
+    description: str = ""
+    pack_name: str = ""
+
+    # Approval configuration (for ESCALATE action)
+    approval_config: dict[str, Any] = field(default_factory=dict)
+
+
+@dataclass
+class CheckContext:
+    """Context passed to rule evaluation."""
+
+    hook: LifecycleHook
+    agent_name: str
+    runtime_id: str
+    trace_id: str
+
+    # Content fields (populated based on hook)
+    agent_input: str = ""
+    agent_output: str = ""
+    model_input: str = ""
+    model_output: str = ""
+    model_name: str = (
+        ""  # LLM model name (e.g., "gpt-4", "claude-3-opus") - available at agent start
+    )
+    tool_name: str = ""
+    tool_args: dict[str, Any] = field(default_factory=dict)
+    tool_result: str = ""
+    messages: list[dict[str, Any]] = field(default_factory=list)
+
+    # Session state
+    session_state: dict[str, Any] = field(default_factory=dict)
+    metadata: dict[str, Any] = field(default_factory=dict)
+
+    # Ring level (privilege level: 0=system, 1=admin, 2=user, 3=untrusted)
+    ring: int = 2
+
+
+@dataclass
+class PolicyPack:
+    """A collection of rules for a compliance standard."""
+
+    name: str
+    version: str
+    description: str
+    rules: list[Rule]
+    enabled: bool = True
+
+
+@dataclass
+class PolicyIndex:
+    """Index of all loaded policy packs and rules."""
+
+    packs: dict[str, PolicyPack] = field(default_factory=dict)
+    _rules_by_id: dict[str, Rule] = field(default_factory=dict)
+    _rules_by_hook: dict[LifecycleHook, list[Rule]] = field(default_factory=dict)
+
+    def add_pack(self, pack: PolicyPack) -> None:
+        """Add a policy pack to the index."""
+        self.packs[pack.name] = pack
+        for rule in pack.rules:
+            rule.pack_name = pack.name
+            self._rules_by_id[rule.rule_id] = rule
+            if rule.hook not in self._rules_by_hook:
+                self._rules_by_hook[rule.hook] = []
+            self._rules_by_hook[rule.hook].append(rule)
+
+    def get_rule(self, rule_id: str) -> Rule | None:
+        """Get a rule by ID."""
+        return self._rules_by_id.get(rule_id)
+
+    def get_rules_for_hook(self, hook: LifecycleHook) -> list[Rule]:
+        """Get all rules for a lifecycle hook."""
+        return self._rules_by_hook.get(hook, [])
+
+    def get_rules_for_pack(self, pack_name: str) -> list[Rule]:
+        """Get all rules for a pack."""
+        pack = self.packs.get(pack_name)
+        return pack.rules if pack else []
+
+    @property
+    def pack_names(self) -> list[str]:
+        """Get all pack names."""
+        return list(self.packs.keys())
+
+    @property
+    def total_rules(self) -> int:
+        """Get total number of rules."""
+        return len(self._rules_by_id)
+
+    @property
+    def all_rules(self) -> list[Rule]:
+        """Get all rules."""
+        return list(self._rules_by_id.values())

tests/conftest.py

@@ -17,3 +17,25 @@ def temp_dir() -> Generator[str, None, None]:
     """Provide a temporary directory for test files."""
     with tempfile.TemporaryDirectory() as tmp_dir:
         yield tmp_dir
+
+
+@pytest.fixture(autouse=True)
+def _reset_governance_process_state() -> Generator[None, None, None]:
+    """Clear process-level governance state around every test.
+
+    The native governance layer keeps two pieces of state at module scope:
+    the conversational/autonomous selector consumed by the policy fetch,
+    and the memoized job-context. Both are stable per process in
+    production but leak across tests when not reset, masking ordering
+    bugs and producing flakes.
+    """
+    from uipath.runtime.governance.native.backend_client import (
+        resolve_job_context,
+        set_agent_conversational,
+    )
+
+    set_agent_conversational(None)
+    resolve_job_context.cache_clear()
+    yield
+    set_agent_conversational(None)
+    resolve_job_context.cache_clear()

tests/test_enforcement_mode_default.py

@@ -0,0 +1,87 @@
+"""Tests for the default enforcement-mode resolution.
+
+The default is :attr:`EnforcementMode.AUDIT` so the wrapper attaches at
+runtime construction and the background policy fetch can run. If the
+backend later returns ``disabled``, ``set_enforcement_mode`` flips the
+cache and ``evaluate()`` short-circuits per-call.
+
+Resolution order (per :func:`get_enforcement_mode`):
+1. Previously-cached programmatic value (set via ``set_enforcement_mode``).
+2. ``UIPATH_GOVERNANCE_MODE`` env var.
+3. Default ``AUDIT``.
+"""
+
+from __future__ import annotations
+
+import pytest
+
+from uipath.runtime.governance.config import (
+    EnforcementMode,
+    get_enforcement_mode,
+    reset_enforcement_mode,
+    set_enforcement_mode,
+)
+
+
+@pytest.fixture(autouse=True)
+def _isolate_mode(monkeypatch: pytest.MonkeyPatch):
+    """Each test starts from a clean module-state slate."""
+    monkeypatch.delenv("UIPATH_GOVERNANCE_MODE", raising=False)
+    reset_enforcement_mode()
+    yield
+    reset_enforcement_mode()
+
+
+def test_default_mode_is_audit() -> None:
+    """No programmatic mode + no env var → AUDIT.
+
+    AUDIT is the default so the wrapper attaches and the background
+    policy fetch can run. The backend can flip the cache to DISABLED
+    on fetch when the tenant has no policies.
+    """
+    assert get_enforcement_mode() is EnforcementMode.AUDIT
+
+
+def test_env_var_disabled_wins_over_default(monkeypatch: pytest.MonkeyPatch) -> None:
+    """Developer override via env var still works."""
+    monkeypatch.setenv("UIPATH_GOVERNANCE_MODE", "disabled")
+    reset_enforcement_mode()  # clear cached default
+    assert get_enforcement_mode() is EnforcementMode.DISABLED
+
+
+def test_env_var_enforce_wins_over_default(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setenv("UIPATH_GOVERNANCE_MODE", "enforce")
+    reset_enforcement_mode()
+    assert get_enforcement_mode() is EnforcementMode.ENFORCE
+
+
+def test_invalid_env_var_falls_back_to_audit(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.setenv("UIPATH_GOVERNANCE_MODE", "garbage-value")
+    reset_enforcement_mode()
+    assert get_enforcement_mode() is EnforcementMode.AUDIT
+
+
+def test_programmatic_set_wins_over_env_and_default(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """The policy loader's ``set_enforcement_mode`` call is canonical."""
+    monkeypatch.setenv("UIPATH_GOVERNANCE_MODE", "audit")
+    set_enforcement_mode(EnforcementMode.ENFORCE)
+    assert get_enforcement_mode() is EnforcementMode.ENFORCE
+
+
+def test_reset_returns_to_default() -> None:
+    """``reset_enforcement_mode`` clears the cache so the default re-applies."""
+    set_enforcement_mode(EnforcementMode.ENFORCE)
+    assert get_enforcement_mode() is EnforcementMode.ENFORCE
+    reset_enforcement_mode()
+    assert get_enforcement_mode() is EnforcementMode.AUDIT
+
+
+def test_audit_mode_is_cached_after_first_read() -> None:
+    """First call computes; subsequent calls hit the cache."""
+    assert get_enforcement_mode() is EnforcementMode.AUDIT
+    # A second call returns the same instance — the cache survives.
+    assert get_enforcement_mode() is EnforcementMode.AUDIT