fix: block_on_review switch, dead code removal, \n in regex

Author: lll-peanut

trpc_agent_sdk/tools/safety/_bash_scanner.py

@@ -30,7 +30,6 @@ class of false positives where a "dangerous" pattern appears inside a string
 from typing import Any
 from typing import Dict
 from typing import List
-from typing import Optional
 from typing import Set
 
 # ---------------------------------------------------------------------------
@@ -353,10 +352,10 @@ def _scan_lines(self) -> None:
     # ------------------------------------------------------------------
 
     def _check_rm(self, line_no: int, raw_line: str, tokens: List[str], args: List[str]) -> None:
-        """Check for recursive-delete.  Flags both ``rm -rf`` and ``rm -r -f``."""
-        all_tokens = set(tokens)
-        # Parse short flags character-by-character (avoids substring false
-        # positives: "-i" has no "r", "-v" has no "f", etc.).
+        """Check for recursive-delete.  Flags both ``rm -rf`` and ``rm -r -f``.
+
+        Parses short flags character-by-character (avoids substring false
+        positives: "-i" has no "r", "-v" has no "f", etc.)."""
         short_flags: set[str] = set()
         has_long_recursive = False
         has_long_force = False

trpc_agent_sdk/tools/safety/_rules.py

@@ -733,7 +733,7 @@ def __call__(
             r"ProcessPoolExecutor\s*\(.*max_workers\s*=\s*(\d+)",
             r"concurrent\.futures",
             r"multiprocessing\.Pool\s*\(.*processes\s*=\s*(\d+)",
-            r"&[\s\n]*done",
+            r"&[\s]*done",
         ]
         for conc_pat in conc_patterns:
             for line_no, line_text in _find_lines(script, conc_pat, script_type=scan_input.script_type):

trpc_agent_sdk/tools/safety/_safety_filter.py

@@ -67,6 +67,10 @@ class ToolSafetyFilter(BaseFilter):
                         events are only emitted via the logger.
         block_on_deny: If True (default), the filter prevents execution when
                        the decision is DENY.
+        block_on_review: If True, also block on NEEDS_HUMAN_REVIEW.
+                         Defaults to False — callers should check
+                         ``rsp.safety_report`` to implement a human-
+                         review gate.
     """
 
     def __init__(
@@ -75,12 +79,14 @@ def __init__(
         policy: Optional[SafetyPolicy] = None,
         audit_log_path: Optional[str] = None,
         block_on_deny: bool = True,
+        block_on_review: bool = False,
     ) -> None:
         super().__init__()
         self._policy = policy or get_policy()
         self._scanner = SafetyScanner(self._policy)
         self._audit = AuditLogger(audit_log_path)
         self._block_on_deny = block_on_deny
+        self._block_on_review = block_on_review
 
         # Identify ourselves within the filter chain
         from trpc_agent_sdk.abc import FilterType
@@ -137,9 +143,10 @@ async def _before(self, ctx: AgentContext, req: Any, rsp: FilterResult) -> None:
                 tool_name,
                 report.summary,
             )
-            # Still allow by default — the caller should check the report.
-            # Set a readable attribute on the result so downstream can decide.
             setattr(rsp, "safety_report", report)
+            if self._block_on_review:
+                rsp.error = ToolSafetyDeniedError(report)
+                rsp.is_continue = False
 
         else:
             logger.debug("ToolSafetyFilter allowed tool '%s'.", tool_name)
@@ -198,7 +205,6 @@ def _collect(val: Any) -> None:
         _collect(getattr(req, "script_content"))
 
     return "\n".join(parts) if parts else None
-    return None
 
 
 def _guess_script_type(req: Any, script: str) -> ScriptType:

trpc_agent_sdk/tools/safety/_scanner.py

@@ -164,8 +164,8 @@ def scan(self, scan_input: SafetyScanInput) -> SafetyScanReport:
                 decision=Decision.DENY,
                 risk_level=RiskLevel.CRITICAL if blocklist_hit else RiskLevel.HIGH,
                 findings=oversized_findings,
-                summary=f"{oversized_reason}." +
-                (" Blocklist pattern matched — denied." if blocklist_hit else " Denied for safety."),
+                summary=(f"{oversized_reason}." +
+                         (" Blocklist pattern matched — denied." if blocklist_hit else " Denied for safety.")),
                 scan_duration_ms=round(duration_ms, 2),
                 policy_version=self._policy.content_hash,
                 sanitized=False,
@@ -397,9 +397,10 @@ def _scan_python_ast(self, script: str, scan_input: SafetyScanInput) -> List[Saf
                             RiskCategory.PROCESS_AND_SYSTEM,
                             risk,
                             f.evidence,
-                            f"AST: {'privilege escalation' if f.extra.get('risk') == 'privilege' else 'process execution'} via {f.canonical_name}",
-                            "Avoid spawning child processes in agent tools." if f.extra.get("risk") != "privilege" else
-                            "Privilege escalation is not allowed in tool scripts.",
+                            ("AST: privilege escalation via " if f.extra.get("risk") == "privilege" else
+                             "AST: process execution via ") + f.canonical_name,
+                            ("Privilege escalation is not allowed in tool scripts." if f.extra.get("risk")
+                             == "privilege" else "Avoid spawning child processes in agent tools."),
                             f.line_number,
                             f.canonical_name,
                         ))