feat(examples): add governance engine and skill_run tool filter

Author: DNKYr

examples/skills_code_review_agent/filters_cr/__init__.py

@@ -0,0 +1,6 @@
+# Tencent is pleased to support the open source community by making tRPC-Agent-Python available.
+#
+# Copyright (C) 2026 Tencent. All rights reserved.
+#
+# tRPC-Agent-Python is licensed under Apache-2.0.
+"""Code review tool filters package."""

examples/skills_code_review_agent/filters_cr/governance_filter.py

@@ -0,0 +1,38 @@
+# Tencent is pleased to support the open source community by making tRPC-Agent-Python available.
+#
+# Copyright (C) 2026 Tencent. All rights reserved.
+#
+# tRPC-Agent-Python is licensed under Apache-2.0.
+"""Tool-level filter guarding LLM-initiated skill_run calls."""
+from trpc_agent_sdk.abc import FilterType
+from trpc_agent_sdk.filter import BaseFilter
+
+from review.governance import GovernanceEngine
+
+
+class GovernanceToolFilter(BaseFilter):
+    """Blocks skill_run commands the GovernanceEngine does not allow."""
+
+    def __init__(self, engine: GovernanceEngine, on_event=None):
+        super().__init__()
+        self._type = FilterType.TOOL
+        self._name = "cr_governance"
+        self._engine = engine
+        self._on_event = on_event
+
+    async def _before(self, ctx, req, rsp):
+        command = ""
+        if isinstance(req, dict):
+            command = str(req.get("command", "") or "")
+        if not command:
+            return
+        decision = self._engine.check_command(command)
+        if self._on_event is not None:
+            self._on_event(decision)
+        if decision.decision != "allow":
+            rsp.rsp = {
+                "error": f"blocked by governance filter ({decision.rule})",
+                "decision": decision.decision,
+                "reason": decision.reason,
+            }
+            rsp.is_continue = False

examples/skills_code_review_agent/review/governance.py

@@ -0,0 +1,87 @@
+# Tencent is pleased to support the open source community by making tRPC-Agent-Python available.
+#
+# Copyright (C) 2026 Tencent. All rights reserved.
+#
+# tRPC-Agent-Python is licensed under Apache-2.0.
+"""Governance policy: script allowlist, forbidden paths, network policy,
+risk heuristics and execution budget."""
+import posixpath
+from dataclasses import dataclass, field
+
+DEFAULT_ALLOWED_SCRIPTS = (
+    "parse_diff.py",
+    "check_security.py",
+    "check_async_leak.py",
+    "check_db_lifecycle.py",
+    "check_tests_missing.py",
+    "check_secrets.py",
+)
+
+NETWORK_TOOLS = frozenset({"curl", "wget", "pip", "pip3", "ssh", "scp", "nc", "git", "apt", "apt-get"})
+RISK_TOKENS = frozenset({"sudo", "docker", "chmod", "chown", "mount", "rm", "mkfs"})
+PYTHON_EXECUTABLES = frozenset({"python", "python3"})
+
+
+@dataclass
+class GovernanceDecision:
+    """Outcome of one governance check."""
+
+    target: str
+    decision: str  # allow | deny | needs_human_review
+    rule: str = ""
+    reason: str = ""
+
+
+@dataclass
+class GovernanceEngine:
+    """Stateful policy engine; deny/needs_human_review must never reach the sandbox."""
+
+    allowed_scripts: tuple = DEFAULT_ALLOWED_SCRIPTS
+    max_runs: int = 20
+    max_sandbox_seconds: float = 300.0
+    _runs: int = field(default=0, init=False)
+    _elapsed: float = field(default=0.0, init=False)
+
+    def record_run(self, duration_s: float) -> None:
+        self._runs += 1
+        self._elapsed += max(0.0, duration_s)
+
+    def _check_budget(self, target: str):
+        if self._runs >= self.max_runs or self._elapsed >= self.max_sandbox_seconds:
+            return GovernanceDecision(target, "deny", "budget_exceeded",
+                                      f"budget: {self._runs} runs / {self._elapsed:.1f}s used")
+        return None
+
+    @staticmethod
+    def _check_paths(target: str, argv):
+        for arg in argv:
+            if arg.startswith("/") or arg.startswith("~") or ".." in arg:
+                return GovernanceDecision(target, "deny", "forbidden_path",
+                                          f"path escapes workspace: {arg!r}")
+        return None
+
+    def check_script(self, script: str, argv) -> GovernanceDecision:
+        target = f"{script} {' '.join(argv)}".strip()
+        blocked = self._check_budget(target) or self._check_paths(target, argv)
+        if blocked:
+            return blocked
+        if posixpath.basename(script) not in self.allowed_scripts:
+            return GovernanceDecision(target, "deny", "script_allowlist",
+                                      f"script {script!r} is not allowlisted")
+        return GovernanceDecision(target, "allow")
+
+    def check_command(self, command: str) -> GovernanceDecision:
+        tokens = command.split()
+        if not tokens:
+            return GovernanceDecision(command, "deny", "empty_command", "empty command")
+        head = posixpath.basename(tokens[0])
+        if head in NETWORK_TOOLS or any(posixpath.basename(t) in NETWORK_TOOLS for t in tokens):
+            return GovernanceDecision(command, "deny", "network_policy",
+                                      "non-whitelisted network access")
+        if head in RISK_TOKENS or any(posixpath.basename(t) in RISK_TOKENS for t in tokens):
+            return GovernanceDecision(command, "needs_human_review", "risk_command",
+                                      "high-risk command requires human review")
+        if head in PYTHON_EXECUTABLES and len(tokens) >= 2:
+            return self.check_script(tokens[1], tokens[2:])
+        return GovernanceDecision(command, "needs_human_review", "unknown_command",
+                                  f"executable {head!r} is not recognized")

examples/skills_code_review_agent/tests/test_governance.py

@@ -0,0 +1,78 @@
+# Tencent is pleased to support the open source community by making tRPC-Agent-Python available.
+#
+# Copyright (C) 2026 Tencent. All rights reserved.
+#
+# tRPC-Agent-Python is licensed under Apache-2.0.
+"""Tests for governance decisions and the tool filter."""
+from trpc_agent_sdk.abc import FilterResult
+
+from review.governance import GovernanceEngine
+from filters_cr.governance_filter import GovernanceToolFilter
+
+
+def test_allowlisted_script_allowed():
+    eng = GovernanceEngine()
+    d = eng.check_script("check_security.py", ["work/inputs/changes.diff"])
+    assert d.decision == "allow"
+
+
+def test_unknown_script_denied():
+    d = GovernanceEngine().check_script("steal_data.py", ["work/inputs/changes.diff"])
+    assert d.decision == "deny"
+    assert d.rule == "script_allowlist"
+
+
+def test_forbidden_path_denied():
+    eng = GovernanceEngine()
+    assert eng.check_script("check_security.py", ["/etc/passwd"]).decision == "deny"
+    assert eng.check_script("check_security.py", ["../../secrets"]).decision == "deny"
+
+
+def test_budget_exceeded_denied():
+    eng = GovernanceEngine(max_runs=1)
+    eng.record_run(0.1)
+    d = eng.check_script("check_security.py", ["work/inputs/changes.diff"])
+    assert d.decision == "deny"
+    assert d.rule == "budget_exceeded"
+
+
+def test_command_network_tool_denied():
+    d = GovernanceEngine().check_command("curl http://evil.example.com")
+    assert d.decision == "deny"
+    assert d.rule == "network_policy"
+
+
+def test_command_risk_token_needs_review():
+    d = GovernanceEngine().check_command("sudo rm -rf /")
+    assert d.decision == "needs_human_review"
+
+
+def test_command_allowlisted_python_script_allowed():
+    d = GovernanceEngine().check_command(
+        "python3 skills/code-review/scripts/check_security.py work/inputs/changes.diff")
+    assert d.decision == "allow"
+
+
+def test_command_unknown_executable_needs_review():
+    d = GovernanceEngine().check_command("make all")
+    assert d.decision == "needs_human_review"
+
+
+async def test_filter_blocks_denied_command():
+    events = []
+    filt = GovernanceToolFilter(GovernanceEngine(), on_event=events.append)
+    rsp = FilterResult()
+    await filt._before(None, {"skill": "code-review", "command": "curl http://x"}, rsp)
+    assert rsp.is_continue is False
+    assert "blocked" in str(rsp.rsp)
+    assert events and events[0].decision == "deny"
+
+
+async def test_filter_allows_good_command():
+    filt = GovernanceToolFilter(GovernanceEngine())
+    rsp = FilterResult()
+    await filt._before(None, {
+        "skill": "code-review",
+        "command": "python3 skills/code-review/scripts/parse_diff.py work/inputs/changes.diff",
+    }, rsp)
+    assert rsp.is_continue is True