style: restore yapf-formatted style for CI formatting check

Revert multi-line formatting changes introduced in d8b9cc7 that broke the yapf --diff CI check. Keep logical fixes (unused import removal, redundant isinstance deletion, policy validation, isinstance simplification) but restore original yapf-approved formatting for should_block, blocked, join_names check, is_process_call, decode-exec check, _ENV_SECRET_KEYS, and environ subscript.

Author: Violet2314

trpc_agent_sdk/safety/_filter.py

@@ -76,10 +76,8 @@ async def _before(self, ctx: Any, req: Any, rsp: FilterResult) -> None:
         )
         report = self.scanner.scan(scan_input)
 
-        should_block = (
-            report.decision == Decision.DENY
-            or (report.decision == Decision.NEEDS_HUMAN_REVIEW and self._block_on_review)
-        )
+        should_block = report.decision == Decision.DENY or (report.decision == Decision.NEEDS_HUMAN_REVIEW
+                                                            and self._block_on_review)
         self.audit.log(report, intercepted=should_block)
 
         if should_block:

trpc_agent_sdk/safety/_rules.py

@@ -366,8 +366,8 @@ def _collect_sensitive_path_vars(
         text = ""
         if isinstance(value, ast.Call):
             cname = resolve_name(value.func, aliases).lower()
-            if (cname in join_names or cname.endswith(".join")
-                    or cname.endswith(".joinpath") or cname.endswith(".expanduser")):
+            if cname in join_names or cname.endswith(".join") or cname.endswith(".joinpath") or cname.endswith(
+                    ".expanduser"):
                 text = path_expr_text(value)
             else:
                 text = path_expr_text(value)
@@ -736,15 +736,10 @@ def _check_python(self, scan_input: ScanInput, policy: PolicyConfig) -> list[Saf
 
         for node, name in iter_python_calls(tree, aliases):
             lname = name.lower()
-            is_process_call = (
-                lname in _PY_PROCESS_CALLS
-                or lname in {c.lower() for c in _PY_PROCESS_CALLS}
-                or any(
+            if lname in _PY_PROCESS_CALLS or any(
                     lname.endswith("." + c.split(".")[-1]) and c.split(".")[0] in lname
-                    for c in _PY_PROCESS_CALLS
-                )
-            )
-            if is_process_call:
+                    for c in _PY_PROCESS_CALLS) or lname in {c.lower()
+                                                             for c in _PY_PROCESS_CALLS}:
                 shell_true = _has_shell_true(node)
                 findings.append(
                     self._finding(
@@ -897,9 +892,8 @@ def _check_bash(self, scan_input: ScanInput, policy: PolicyConfig) -> list[Safet
                         level=RiskLevel.CRITICAL,
                         message="Use of eval in bash",
                     ))
-            if (_DECODE_EXEC_BASH.search(line)
-                    or (re.search(r"\|\s*(sh|bash|zsh)\b", line)
-                        and re.search(r"base64|xxd|openssl", line, re.IGNORECASE))):
+            if _DECODE_EXEC_BASH.search(line) or re.search(r"\|\s*(sh|bash|zsh)\b", line) and re.search(
+                    r"base64|xxd|openssl", line, re.IGNORECASE):
                 findings.append(
                     self._finding(
                         line,
@@ -1272,9 +1266,8 @@ def _const_int(node: ast.AST | None) -> int | None:
     "httpx.post",
 }
 
-_ENV_SECRET_KEYS = re.compile(
-    r"(?i)(API[_-]?KEY|TOKEN|SECRET|PASSWORD|PASSWD|PRIVATE[_-]?KEY|"
-    r"ACCESS[_-]?KEY|OPENAI|ANTHROPIC|AWS_SECRET|GITHUB_TOKEN|GH_TOKEN)")
+_ENV_SECRET_KEYS = re.compile(r"(?i)(API[_-]?KEY|TOKEN|SECRET|PASSWORD|PASSWD|PRIVATE[_-]?KEY|"
+                              r"ACCESS[_-]?KEY|OPENAI|ANTHROPIC|AWS_SECRET|GITHUB_TOKEN|GH_TOKEN)")
 
 
 class SecretLeakRule(SafetyRule):
@@ -1376,11 +1369,8 @@ def _check_python(self, scan_input: ScanInput, policy: PolicyConfig) -> list[Saf
         # Direct print(os.environ["X"])
         for node in ast.walk(tree):
             if isinstance(node, ast.Subscript) and _is_environ_subscript(node, aliases):
-                # ast.Subscript always has .slice in py>=3.9; fall back to None
-                # for older interpreters or unusual AST shapes.
-                key = getattr(getattr(node, "slice", None), "value", None)
-                key_str = get_string_literal(key) if key is not None else None
-                if key_str is None or _ENV_SECRET_KEYS.search(str(key_str if key_str else "SECRET")):
+                key = get_string_literal(node.slice) if hasattr(node, "slice") else None
+                if key is None or _ENV_SECRET_KEYS.search(str(key) if key else "SECRET"):
                     # Flag any environ subscript used as expression; severity HIGH
                     findings.append(
                         self._finding(

trpc_agent_sdk/safety/_scanner.py

@@ -161,10 +161,8 @@ def scan(self, scan_input: ScanInput) -> SafetyReport:
 
         # blocked means "would be intercepted under this policy": DENY always,
         # and NEEDS_HUMAN_REVIEW when policy.block_on_review is enabled.
-        blocked = (
-            decision == Decision.DENY
-            or (decision == Decision.NEEDS_HUMAN_REVIEW and self.policy.block_on_review)
-        )
+        blocked = (decision == Decision.DENY
+                   or (decision == Decision.NEEDS_HUMAN_REVIEW and self.policy.block_on_review))
         return SafetyReport(
             decision=decision,
             risk_level=agg_level,