examples: add Filter policy gate to code-review agent

Slice 3 (requirement 7). pipeline/policy.py::ReviewPolicy decides
allow / deny / needs_human_review for a sandbox action based on the
command (high-risk patterns), touched paths (forbidden roots),
network hosts (allowlist) and budget. It is enforced at two sites
sharing the policy:

the deterministic sandbox gate (pipeline/sandbox.py): a denied or
human-review action is never launched; a blocked SandboxRunResult is
recorded and surfaced in the report's Filter-interception section
(completing requirement 8) and in monitoring.block_count.
agent/filter.py::ReviewGuardFilter: a TOOL-scoped BaseFilter attached
on the review tool that refuses a guarded tool call via
FilterResult(is_continue=False).

Adds tests for the policy decisions, that a denied action never reaches
the sandbox, the guard filter, and the report's block section.

Updates #92

RELEASE NOTES: NONE

Author: Acture

examples/skills_code_review_agent/README.md

@@ -83,5 +83,13 @@ self-test, CLI, the fake-model agent loop (`run_agent.py`), and **sandbox execut
 each run; `--runtime container` runs them in a Docker workspace — see `skills/code-review/Dockerfile`
 — and is skipped in tests when Docker is absent). Baseline secret redaction is in place.
 
-Planned follow-up slices: the tool-level Filter gate (criterion 7), redaction hardening to ≥95%
-(criterion 5), and OpenTelemetry metrics wiring (criterion 9).
+The **Filter gate** (criterion 7) is in place: `pipeline/policy.py::ReviewPolicy` decides
+allow / deny / needs-human-review for a sandbox action (high-risk command, forbidden path,
+non-whitelisted network, over-budget). It is enforced at two sites sharing that policy — the
+deterministic sandbox gate (a denied action never launches; the block is recorded and surfaced in
+the report's Filter-interception section) and the framework `agent/filter.py::ReviewGuardFilter`
+(TOOL-scoped, attached on the review tool).
+
+Planned follow-up slices: redaction hardening to ≥95% (criterion 5) and OpenTelemetry metrics wiring
+(criterion 9). Note: the default runtime is `inprocess` for fast dry-runs — pass `--runtime local`
+or `--runtime container` to exercise the sandbox path.

examples/skills_code_review_agent/agent/filter.py

@@ -0,0 +1,47 @@
+# 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.
+
+"""ReviewGuardFilter — the framework enforcement site for the review policy (issue #92, req 7).
+
+A TOOL-scoped filter (``register_tool_filter``): it inspects a tool call's args before the tool
+runs and refuses high-risk sandbox actions, so ``deny`` / ``needs_human_review`` never reach
+execution. Attach it on the tool (``FunctionTool(fn, filters_name=["review_guard"])``), NOT on the
+agent — the agent resolves ``filters_name`` in the AGENT namespace and would raise.
+
+It shares its decision logic with the deterministic sandbox gate via ``pipeline.policy.ReviewPolicy``.
+"""
+from __future__ import annotations
+
+from typing import Any
+
+from trpc_agent_sdk.filter import BaseFilter, FilterResult, register_tool_filter
+
+from pipeline.policy import ReviewPolicy
+
+_GUARDED_ARG_KEYS = ("command", "script", "cmd")
+
+
+@register_tool_filter("review_guard")
+class ReviewGuardFilter(BaseFilter):
+    """Blocks a guarded tool call whose command the policy denies or flags for human review."""
+
+    policy = ReviewPolicy()
+
+    def _command(self, req: Any) -> str:
+        if isinstance(req, dict):
+            for key in _GUARDED_ARG_KEYS:
+                if req.get(key):
+                    return str(req[key])
+        return ""
+
+    async def _before(self, ctx: Any, req: Any, rsp: FilterResult) -> None:
+        command = self._command(req)
+        if not command:
+            return  # nothing risky to gate (e.g. review_code(diff_text=...))
+        decision = self.policy.evaluate(command=command)
+        if not decision.allowed:
+            rsp.is_continue = False
+            rsp.error = PermissionError(f"review_guard blocked ({decision.category}): {decision.reason}")

examples/skills_code_review_agent/agent/tools.py

@@ -3,7 +3,6 @@
 # Copyright (C) 2026 Tencent. All rights reserved.
 #
 # tRPC-Agent-Python is licensed under Apache-2.0.
-
 """The review tool the agent calls — a thin wrapper over the deterministic pipeline."""
 from __future__ import annotations
 
@@ -13,6 +12,8 @@
 
 from pipeline.engine import run_review
 
+from . import filter as _guard  # noqa: F401 - importing registers the "review_guard" tool filter
+
 
 def review_code(diff_text: str) -> dict[str, Any]:
     """Run the code-review pipeline on a unified diff and return a findings summary.
@@ -29,4 +30,5 @@ def review_code(diff_text: str) -> dict[str, Any]:
 
 
 def build_review_tool() -> FunctionTool:
-    return FunctionTool(review_code)
+    # The guard is TOOL-scoped: attach on the tool, not on the agent.
+    return FunctionTool(review_code, filters_name=["review_guard"])

examples/skills_code_review_agent/pipeline/engine.py

@@ -24,6 +24,7 @@
 
 from . import diff_parser, report as report_mod, scanners
 from .dedup import dedup_and_denoise
+from .policy import ReviewPolicy
 from .types import DiffSummary, Finding, ReviewReport
 
 
@@ -60,6 +61,7 @@ def run_review(
     runtime: str = "inprocess",
     sandbox_timeout: float | None = None,
     max_output_bytes: int | None = None,
+    policy: ReviewPolicy | None = None,
     warn_threshold: float | None = None,
     review_threshold: float | None = None,
 ) -> ReviewResult:
@@ -92,9 +94,10 @@ def run_review(
         raw, run = sandbox_mod.run_local(
             scan_dir,
             timeout=sandbox_timeout if sandbox_timeout is not None else sandbox_mod.DEFAULT_TIMEOUT_SEC,
-            max_bytes=max_output_bytes if max_output_bytes is not None else sandbox_mod.MAX_OUTPUT_BYTES)
+            max_bytes=max_output_bytes if max_output_bytes is not None else sandbox_mod.MAX_OUTPUT_BYTES,
+            policy=policy if policy is not None else ReviewPolicy())
         sandbox_runs = [run]
-        if run.timed_out or run.exit_code not in (0, 1):  # 1 = scanners found issues (normal)
+        if run.timed_out or (not run.blocked and run.exit_code not in (0, 1)):  # 1 = issues found (normal)
             exception_dist["sandbox_failure"] = exception_dist.get("sandbox_failure", 0) + 1
     else:  # "inprocess"
         try:
@@ -124,17 +127,27 @@ def _assemble(task_id, summary, raw, sandbox_runs, source_type, source_ref, star
     for f in active:
         severity_dist[f.severity] = severity_dist.get(f.severity, 0) + 1
 
+    filter_blocks = [{
+        "script": r.script,
+        "reason": r.block_reason,
+        "category": r.block_category
+    } for r in sandbox_runs if r.blocked]
+
     monitoring = {
         "total_sec": round(time.monotonic() - started, 3),
         "sandbox_sec": round(sum(r.duration_sec for r in sandbox_runs), 3),
         "tool_calls": len(scanners.ADAPTERS),
-        "block_count": 0,  # populated in slice 3 (Filter gate)
+        "block_count": len(filter_blocks),
         "finding_count": len(active),
         "severity_dist": severity_dist,
         "exception_dist": exception_dist,
     }
 
-    report = report_mod.build_report(task_id, findings, sandbox_runs=sandbox_runs, monitoring=monitoring)
+    report = report_mod.build_report(task_id,
+                                     findings,
+                                     sandbox_runs=sandbox_runs,
+                                     filter_blocks=filter_blocks,
+                                     monitoring=monitoring)
     return ReviewResult(task_id=task_id,
                         report=report,
                         findings=findings,

examples/skills_code_review_agent/pipeline/policy.py

@@ -0,0 +1,83 @@
+# 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.
+
+"""Review policy — the Filter decision logic (issue #92, requirement 7 & 8).
+
+Shared by two enforcement sites (plan decision #5): the framework ``ReviewGuardFilter`` on the agent
+path, and a direct gate in the sandbox runner on the deterministic path. Both call ``evaluate`` and
+must refuse to execute anything that comes back ``deny`` or ``needs_human_review``.
+"""
+from __future__ import annotations
+
+import os
+import re
+from dataclasses import dataclass
+from typing import Iterable, Literal
+
+Decision = Literal["allow", "deny", "needs_human_review"]
+
+# High-risk command patterns → hard deny.
+_DANGEROUS = [
+    (re.compile(r"\brm\s+-[rfRF]"), "recursive/force delete"),
+    (re.compile(r"\b(mkfs|shred)\b|\bdd\s+if="), "disk-destructive command"),
+    (re.compile(r":\s*\(\s*\)\s*\{.*\}\s*;"), "fork bomb"),
+    (re.compile(r"(curl|wget)\b[^\n|]*\|\s*(sh|bash)"), "pipe-to-shell"),
+    (re.compile(r"\bchmod\s+-?R?\s*777\b"), "world-writable chmod"),
+    (re.compile(r"\bsudo\b"), "privilege escalation"),
+    (re.compile(r">\s*/dev/sd|/etc/passwd|/etc/shadow"), "write to sensitive target"),
+]
+
+# Sensitive roots a review must never touch (temp dirs under /var/folders are intentionally allowed).
+_FORBIDDEN_PATHS = ("/etc", "/root", "/boot", os.path.expanduser("~/.ssh"))
+
+
+@dataclass
+class PolicyDecision:
+    decision: Decision
+    reason: str = ""
+    category: str = "ok"  # script | path | network | budget | ok
+
+    @property
+    def allowed(self) -> bool:
+        return self.decision == "allow"
+
+    def as_block(self, *, script: str = "") -> dict:
+        return {"script": script, "decision": self.decision, "reason": self.reason, "category": self.category}
+
+
+class ReviewPolicy:
+    """Decides whether a sandbox action may run. Deny/needs_human_review must NOT reach the sandbox."""
+
+    def __init__(self, network_allowlist: Iterable[str] | None = None, max_budget_sec: float = 120.0) -> None:
+        self.network_allowlist = set(network_allowlist or ())
+        self.max_budget_sec = max_budget_sec
+
+    def evaluate(
+            self,
+            *,
+            command: str = "",
+            touched_paths: Iterable[str] = (),
+            network_hosts: Iterable[str] = (),
+            budget_sec: float = 0.0,
+    ) -> PolicyDecision:
+        for pat, why in _DANGEROUS:
+            if pat.search(command):
+                return PolicyDecision("deny", f"high-risk command: {why}", "script")
+
+        for p in touched_paths:
+            ap = os.path.abspath(p)
+            if any(ap == fp or ap.startswith(fp + os.sep) for fp in _FORBIDDEN_PATHS):
+                return PolicyDecision("deny", f"forbidden path: {p}", "path")
+
+        unlisted = [h for h in network_hosts if h not in self.network_allowlist]
+        if unlisted:
+            return PolicyDecision("needs_human_review", f"non-whitelisted network: {unlisted}", "network")
+
+        if budget_sec and budget_sec > self.max_budget_sec:
+            return PolicyDecision("needs_human_review", f"over budget: {budget_sec}s > {self.max_budget_sec}s",
+                                  "budget")
+
+        return PolicyDecision("allow")

examples/skills_code_review_agent/pipeline/sandbox.py

@@ -3,7 +3,6 @@
 # Copyright (C) 2026 Tencent. All rights reserved.
 #
 # tRPC-Agent-Python is licensed under Apache-2.0.
-
 """Sandboxed execution of the review scanners (issue #92, requirement 4 & slice 2).
 
 Runs the skill's ``run_checks.py`` outside the review process, with a timeout and an output-size cap,
@@ -24,14 +23,36 @@
 import tempfile
 import time
 from pathlib import Path
+from typing import TYPE_CHECKING
 
 from .types import Finding, SandboxRunResult
 
+if TYPE_CHECKING:
+    from .policy import ReviewPolicy
+
 _SKILL_SCRIPT = Path(__file__).resolve().parents[3] / "skills" / "code-review" / "scripts" / "run_checks.py"
 DEFAULT_TIMEOUT_SEC = 60.0
 MAX_OUTPUT_BYTES = 1_048_576  # 1 MiB per stream
 
 
+def _gate(policy: "ReviewPolicy | None", cmd: list[str], scan_dir: str, timeout: float) -> SandboxRunResult | None:
+    """Return a blocked result if the policy refuses the action, else None (allowed to run)."""
+    if policy is None:
+        return None
+    decision = policy.evaluate(command=" ".join(cmd), touched_paths=[scan_dir], budget_sec=timeout)
+    if decision.allowed:
+        return None
+    return SandboxRunResult(script="run_checks.py",
+                            exit_code=0,
+                            duration_sec=0.0,
+                            timed_out=False,
+                            stdout_bytes=0,
+                            stderr_bytes=0,
+                            blocked=True,
+                            block_reason=decision.reason,
+                            block_category=decision.category)
+
+
 def _truncate(text: str | bytes | None, cap: int) -> tuple[str, int]:
     """Return (possibly-truncated text, original byte length). The cap bounds what we persist."""
     if text is None:
@@ -69,11 +90,20 @@ def run_local(
     *,
     timeout: float = DEFAULT_TIMEOUT_SEC,
     max_bytes: int = MAX_OUTPUT_BYTES,
+    policy: "ReviewPolicy | None" = None,
 ) -> tuple[list[Finding], SandboxRunResult]:
-    """Run the scanners in a subprocess against ``scan_dir``; never raises."""
+    """Run the scanners in a subprocess against ``scan_dir``; never raises.
+
+    If ``policy`` denies the action (or requires human review), the subprocess is NOT launched and a
+    blocked ``SandboxRunResult`` is returned instead (requirement 7).
+    """
     out_file = Path(tempfile.mkdtemp(prefix="cr_out_")) / "findings.json"
     cmd = [sys.executable, str(_SKILL_SCRIPT), "--target", scan_dir, "--out", str(out_file)]
 
+    blocked = _gate(policy, cmd, scan_dir, timeout)
+    if blocked is not None:
+        return [], blocked
+
     started = time.monotonic()
     timed_out = False
     exit_code = 0

tests/examples/test_skills_code_review_agent.py

@@ -168,3 +168,57 @@ def test_sandbox_output_byte_accounting() -> None:
     text, n = _truncate("x" * 5000, 10)
     assert n == 5000  # records the true size
     assert len(text.encode()) <= 10 + len("\n...[truncated]")
+
+
+def test_policy_decisions() -> None:
+    from pipeline.policy import ReviewPolicy
+
+    p = ReviewPolicy()
+    assert p.evaluate(command="rm -rf /tmp/x").decision == "deny"
+    assert p.evaluate(command="python run_checks.py").decision == "allow"
+    assert p.evaluate(command="cat x", touched_paths=["/etc/passwd"]).decision == "deny"
+    assert p.evaluate(command="fetch", network_hosts=["evil.com"]).decision == "needs_human_review"
+
+
+def test_denied_action_never_reaches_sandbox() -> None:
+    from pipeline.policy import ReviewPolicy
+
+    # A policy that refuses everything (tiny budget) must block before execution (requirement 7).
+    result = run_review(diff_text=(_FIXTURES / "0001_insecure.diff").read_text(),
+                        runtime="local",
+                        policy=ReviewPolicy(max_budget_sec=1e-6),
+                        sandbox_timeout=60)
+    run = result.report.sandbox_summary[0]
+    assert run.blocked is True
+    assert run.duration_sec == 0.0  # never executed
+    assert result.report.findings_summary["total"] == 0
+    assert result.report.filter_blocks and result.report.filter_blocks[0]["category"] == "budget"
+    assert result.monitoring["block_count"] == 1
+
+
+@pytest.mark.asyncio
+async def test_guard_filter_blocks_dangerous_command() -> None:
+    from trpc_agent_sdk.filter import FilterResult
+
+    from agent.filter import ReviewGuardFilter
+
+    guard = ReviewGuardFilter()
+    dangerous = FilterResult()
+    await guard._before(None, {"command": "rm -rf /"}, dangerous)
+    assert dangerous.is_continue is False
+
+    safe = FilterResult()
+    await guard._before(None, {"diff_text": "some diff"}, safe)
+    assert safe.is_continue is True  # review_code has no command arg -> passes
+
+
+def test_report_renders_filter_block_section() -> None:
+    from pipeline.policy import ReviewPolicy
+
+    result = run_review(diff_text=(_FIXTURES / "0001_insecure.diff").read_text(),
+                        runtime="local",
+                        policy=ReviewPolicy(max_budget_sec=1e-6),
+                        sandbox_timeout=60)
+    md = report_mod.render_md(result.report)
+    assert "## 4. Filter interception summary" in md
+    assert "over budget" in md