fix: blocklist_commands enforcement, &-split, scanner cache, W605

  - _scanner: blocklist_commands from policy now force DENY (was dead config)
  - _bash_scanner: & added to dispatch_commands separators
  - _scanner: get_scanner() cache invalidated on policy content_hash change
  - _bash_scanner: fix W605 invalid escape sequence in docstring

Author: lll-peanut

trpc_agent_sdk/tools/safety/_bash_scanner.py

@@ -252,14 +252,14 @@ def _dispatch_commands(self, line_no: int, raw_line: str, tokens: List[str]) ->
         """
         seg_start = 0
         for i, t in enumerate(tokens):
-            if t in (";", "&&", "||", "|"):
+            if t in (";", "&&", "||", "|", "&"):
                 self._analyse_one_command_with_dynamic_scan(line_no, raw_line, tokens[seg_start:i])
                 seg_start = i + 1
         if seg_start < len(tokens):
             self._analyse_one_command_with_dynamic_scan(line_no, raw_line, tokens[seg_start:])
 
     def _analyse_one_command_with_dynamic_scan(self, line_no: int, raw_line: str, cmd_tokens: List[str]) -> None:
-        """Analyse a command segment AND scan all tokens for eval/exec inside \$(...)."""
+        r"""Analyse a command segment AND scan all tokens for eval/exec inside \$(...)."""
         self._analyse_one_command(line_no, raw_line, cmd_tokens)
         # Scan for dynamic commands that appear anywhere in the token stream
         # (not just as the head command), so that $(eval "rm -rf /") and

trpc_agent_sdk/tools/safety/_scanner.py

@@ -225,6 +225,24 @@ def scan(self, scan_input: SafetyScanInput) -> SafetyScanReport:
         if decision != Decision.DENY:
             decision, bl_findings = self._check_blocklist_override(script, decision, scan_input.script_type)
             all_findings.extend(bl_findings)
+        # Blocklist commands — also force DENY when a forbidden command literal appears
+        if decision != Decision.DENY and self._policy.blocklist_commands:
+            for cmd in self._policy.blocklist_commands:
+                if re.search(re.escape(cmd), script, re.IGNORECASE):
+                    logger.warning("Blocklist command matched: %s → forcing DENY", cmd)
+                    all_findings.append(
+                        SafetyFinding(
+                            rule_id="FILE-001",
+                            category=RiskCategory.DANGEROUS_FILE_OPS,
+                            risk_level=RiskLevel.CRITICAL,
+                            evidence=cmd,
+                            message=f"Blocklisted command detected: {cmd}",
+                            recommendation="Remove the dangerous command from the script.",
+                            line_number=0,
+                            matched_pattern=cmd,
+                        ))
+                    decision = Decision.DENY
+                    break
 
         # Apply allow-pattern override — allow patterns → allow
         # Only upgrades NEEDS_HUMAN_REVIEW; never overrides DENY (blocklist wins).
@@ -857,10 +875,15 @@ def _redact_evidence(self, findings: list[SafetyFinding]) -> list[SafetyFinding]
 
 
 def get_scanner() -> SafetyScanner:
-    """Return (and cache) the default SafetyScanner instance."""
+    """Return (and cache) the default SafetyScanner instance.
+
+    The cache is invalidated on policy change so that ``reload_policy()``
+    + ``quick_scan()`` sees the updated rules.
+    """
     global _default_scanner  # pylint: disable=global-statement
-    if _default_scanner is None:
-        _default_scanner = SafetyScanner()
+    current_policy = get_policy()
+    if _default_scanner is None or _default_scanner._policy.content_hash != current_policy.content_hash:
+        _default_scanner = SafetyScanner(policy=current_policy)
     return _default_scanner