feat: add strict mode and async policy evaluation

Address review feedback:
- Add strict=True parameter that raises RuntimeError on policy load
  failures instead of silently skipping (for security-critical deployments)
- Offload synchronous PolicyEngine.evaluate to thread pool executor
  via asyncio.run_in_executor to avoid blocking the event loop
- Add 3 new tests (strict mode, non-strict skip, thread pool verification)
- Update sample README with strict mode documentation

Signed-off-by: Imran Siddique <imran.siddique@microsoft.com>

Author: Imran Siddique

contributing/samples/agent_governance/README.md

@@ -61,6 +61,19 @@ If `agentmesh-platform` is not installed, the plugin raises `ImportError`
 at construction time. Pass `fail_open=True` to degrade gracefully instead
 (all calls pass through with a logged warning).
 
+## Strict mode
+
+By default, the plugin skips policy files that fail to parse and logs a
+warning. Pass `strict=True` to raise a `RuntimeError` instead, which is
+recommended when every policy file is security-critical:
+
+```python
+plugin = AgentGovernancePlugin(
+    policy_dir=Path(__file__).parent / "policies",
+    strict=True,  # abort if any policy fails to load
+)
+```
+
 ## Links
 
 - [Agent Governance Toolkit](https://github.com/microsoft/agent-governance-toolkit)

src/google/adk_community/plugins/agent_governance_plugin.py

@@ -23,6 +23,8 @@
 
 from __future__ import annotations
 
+import asyncio
+import functools
 import logging
 from pathlib import Path
 from typing import Any, Optional
@@ -53,10 +55,15 @@ class AgentGovernancePlugin(BasePlugin):
         fail_open: If ``True``, tool calls proceed when ``agentmesh-platform``
             is not installed (logs a warning). If ``False`` (default), raises
             ``ImportError`` at construction time.
+        strict: If ``True``, raises on policy load errors instead of
+            skipping the file. Useful when every policy file is
+            security-critical and a partial load is unacceptable.
+            Defaults to ``False``.
 
     Raises:
         ImportError: If ``agentmesh-platform`` is not installed and
             ``fail_open`` is False.
+        RuntimeError: If ``strict`` is True and a policy file fails to load.
 
     Example::
 
@@ -73,15 +80,16 @@ def __init__(
         policy_dir: str | Path,
         agent_did: str = "did:mesh:adk-agent",
         fail_open: bool = False,
+        strict: bool = False,
     ) -> None:
         super().__init__(name="agent_governance")
         self._policy_dir = Path(policy_dir).resolve()
         self._agent_did = agent_did
         self._engine = None
         self._audit = None
-        self._setup(fail_open=fail_open)
+        self._setup(fail_open=fail_open, strict=strict)
 
-    def _setup(self, *, fail_open: bool) -> None:
+    def _setup(self, *, fail_open: bool, strict: bool) -> None:
         """Initialize AGT policy engine and audit service."""
         try:
             from agentmesh.governance.policy import PolicyEngine
@@ -96,6 +104,10 @@ def _setup(self, *, fail_open: bool) -> None:
                         self._engine.load_yaml(f.read_text())
                         logger.info("Loaded policy: %s", f.name)
                     except Exception as exc:
+                        if strict:
+                            raise RuntimeError(
+                                f"Failed to load policy {f.name}: {exc}"
+                            ) from exc
                         logger.warning("Skipped %s: %s", f.name, exc)
             else:
                 logger.warning(
@@ -138,9 +150,14 @@ async def before_tool_callback(
             "tool_args": tool_args,
         }
 
-        result = self._engine.evaluate(
-            agent_did=self._agent_did,
-            context=context,
+        loop = asyncio.get_running_loop()
+        result = await loop.run_in_executor(
+            None,
+            functools.partial(
+                self._engine.evaluate,
+                agent_did=self._agent_did,
+                context=context,
+            ),
         )
 
         if self._audit:

tests/plugins/test_agent_governance_plugin.py

@@ -297,3 +297,83 @@ def test_custom_agent_did(self, policy_dir: Path):
             agent_did="did:mesh:custom-agent",
         )
         assert plugin._agent_did == "did:mesh:custom-agent"
+
+    def test_strict_mode_raises_on_bad_policy(self, tmp_path: Path):
+        """With strict=True, plugin raises RuntimeError on policy load failure."""
+        _install_fake_agentmesh()
+
+        # Patch _FakePolicyEngine.load_yaml to raise on bad content
+        original_load = _FakePolicyEngine.load_yaml
+
+        def failing_load(self, text):
+            if "INVALID" in text:
+                raise ValueError("invalid YAML syntax")
+            original_load(self, text)
+
+        _FakePolicyEngine.load_yaml = failing_load
+
+        policies = tmp_path / "policies"
+        policies.mkdir()
+        (policies / "bad.yaml").write_text("INVALID policy content")
+
+        try:
+            with pytest.raises(RuntimeError, match="Failed to load policy"):
+                AgentGovernancePlugin(policy_dir=policies, strict=True)
+        finally:
+            _FakePolicyEngine.load_yaml = original_load
+
+    def test_non_strict_skips_bad_policy(self, tmp_path: Path):
+        """Without strict mode, bad policies are skipped with a warning."""
+        _install_fake_agentmesh()
+
+        original_load = _FakePolicyEngine.load_yaml
+
+        def failing_load(self, text):
+            if "INVALID" in text:
+                raise ValueError("invalid YAML syntax")
+            original_load(self, text)
+
+        _FakePolicyEngine.load_yaml = failing_load
+
+        policies = tmp_path / "policies"
+        policies.mkdir()
+        (policies / "good.yaml").write_text("valid policy")
+        (policies / "bad.yaml").write_text("INVALID policy content")
+
+        try:
+            plugin = AgentGovernancePlugin(policy_dir=policies)
+            # Good policy loaded, bad one skipped
+            assert len(plugin._engine._policies) == 1
+        finally:
+            _FakePolicyEngine.load_yaml = original_load
+
+    @pytest.mark.asyncio
+    async def test_evaluate_runs_in_thread_pool(self, policy_dir: Path):
+        """Policy evaluation is offloaded to a thread pool executor."""
+        import asyncio
+
+        _install_fake_agentmesh()
+        plugin = AgentGovernancePlugin(policy_dir=policy_dir)
+
+        # Track which thread evaluate() runs on
+        import threading
+
+        eval_thread = None
+        original_evaluate = plugin._engine.evaluate
+
+        def tracking_evaluate(**kwargs):
+            nonlocal eval_thread
+            eval_thread = threading.current_thread()
+            return original_evaluate(**kwargs)
+
+        plugin._engine.evaluate = tracking_evaluate
+        main_thread = threading.current_thread()
+
+        await plugin.before_tool_callback(
+            tool=_FakeBaseTool("web_search"),
+            tool_args={"query": "test"},
+            tool_context=_FakeToolContext(),
+        )
+
+        assert eval_thread is not None
+        assert eval_thread != main_thread