fix(safety): resolve CI lint failures (yapf + flake8)

CI lint job (YAPF + flake8) failed on the safety package. Fixes:
- Remove unused Decision/RiskLevel imports and unused `norm` local
- Reformat the package with yapf (column_limit=120,
  split_before_logical_operator) to match CI style exactly
- Wrap an over-long ValueError message in _policy._decision

flake8 + yapf now clean on all changed trpc_agent_sdk/*.py; 73 tests pass.

Author: wzxsdf

trpc_agent_sdk/tools/__init__.py

@@ -217,4 +217,5 @@ def __getattr__(name):
 def __dir__():
     return sorted(set(list(globals()) + list(_LAZY_REEXPORTS)))
 
+
 from trpc_agent_sdk.tools import safety  # noqa: F401

trpc_agent_sdk/tools/safety/_audit.py

@@ -81,7 +81,8 @@ def build_audit_record(
         language=language,
         decision=report.decision.name,
         risk_level=report.risk_level.name,
-        rule_ids=sorted({f.rule_id for f in report.findings}),
+        rule_ids=sorted({f.rule_id
+                         for f in report.findings}),
         scan_duration_ms=report.scan_duration_ms,
         sanitized=bool(report.sanitized),
         intercepted=bool(intercepted),

trpc_agent_sdk/tools/safety/_bash_scanner.py

@@ -35,42 +35,39 @@ def scan_bash(policy: Policy, script: str) -> list[Finding]:
 
     def add(rule_id: str, evidence: str, rec: str) -> None:
         meta = rule_meta[rule_id]
-        findings.append(Finding(
-            rule_id=rule_id,
-            risk_level=meta.risk_level,
-            rule_decision=meta.decision,
-            evidence=evidence[:max_ev],
-            recommendation=rec,
-            language="bash",
-        ))
+        findings.append(
+            Finding(
+                rule_id=rule_id,
+                risk_level=meta.risk_level,
+                rule_decision=meta.decision,
+                evidence=evidence[:max_ev],
+                recommendation=rec,
+                language="bash",
+            ))
 
     joined = script
 
     # Recursive delete
-    if re.search(r"\brm\b[^;\n]*-r[f]?", joined) and ("/" == _rm_target(joined) or
-                                                      re.search(r"rm\s+-[rf]+\s+/", joined)):
-        add(R_FS_RECURSIVE_DELETE, "rm -rf against root/system path",
-            "Refuse recursive delete of system paths.")
+    if re.search(r"\brm\b[^;\n]*-r[f]?", joined) and ("/" == _rm_target(joined)
+                                                      or re.search(r"rm\s+-[rf]+\s+/", joined)):
+        add(R_FS_RECURSIVE_DELETE, "rm -rf against root/system path", "Refuse recursive delete of system paths.")
 
     # Fork bomb
     if _FORK_BOMB_RE.search(joined):
         add(R_RES_FORK_BOMB, "fork bomb pattern", "Refuse fork bomb.")
 
     # Dependency install
     if re.search(r"\b(pip|pip3|npm|yarn|apt|apt-get|yum|brew)\s+install\b", joined):
-        add(R_PKG_INSTALL, "dependency install command",
-            "Installing deps changes the runtime environment; review.")
+        add(R_PKG_INSTALL, "dependency install command", "Installing deps changes the runtime environment; review.")
 
     # Privilege escalation
     if re.search(r"\b(sudo|su|doas)\b", joined):
-        add(R_PROC_PRIVILEGE_ESCALATION, "privilege escalation command",
-            "Privilege escalation requires review.")
+        add(R_PROC_PRIVILEGE_ESCALATION, "privilege escalation command", "Privilege escalation requires review.")
 
     # Long sleep (>= policy.max_timeout_seconds)
     for m in _SLEEP_RE.finditer(joined):
         if int(m.group(1)) >= policy.max_timeout_seconds:
-            add(R_RES_LONG_SLEEP, f"sleep {m.group(1)}",
-                f"sleep >= {policy.max_timeout_seconds}s is suspicious.")
+            add(R_RES_LONG_SLEEP, f"sleep {m.group(1)}", f"sleep >= {policy.max_timeout_seconds}s is suspicious.")
             break
 
     # Large write: dd/truncate with GB+ size, or head -c with a huge byte count.
@@ -79,8 +76,7 @@ def add(rule_id: str, evidence: str, rec: str) -> None:
             "Very large file generation; possible disk exhaustion. Review.")
     for m in _HEAD_C_RE.finditer(joined):
         if int(m.group(1)) >= _HUGE_BYTES:
-            add(R_RES_LARGE_WRITE, f"head -c {m.group(1)}",
-                "Very large write; possible disk exhaustion. Review.")
+            add(R_RES_LARGE_WRITE, f"head -c {m.group(1)}", "Very large write; possible disk exhaustion. Review.")
             break
 
     # Shell pipe / bypass
@@ -100,8 +96,7 @@ def add(rule_id: str, evidence: str, rec: str) -> None:
         host = m.group(1).lower()
         root_domain = ".".join(host.split(".")[-2:]) if len(host.split(".")) >= 2 else host
         if root_domain not in policy.whitelisted_domains and host not in policy.whitelisted_domains:
-            add(R_NET_HTTP, f"network egress to {host}",
-                f"{host} is not whitelisted; review or allowlist.")
+            add(R_NET_HTTP, f"network egress to {host}", f"{host} is not whitelisted; review or allowlist.")
 
     return findings
 

trpc_agent_sdk/tools/safety/_code_executor_guard.py

@@ -49,22 +49,18 @@ def _ensure_policy(self) -> Policy:
         return self._policy
 
     @override
-    async def execute_code(self,
-                           invocation_context: InvocationContext,
+    async def execute_code(self, invocation_context: InvocationContext,
                            input_data: CodeExecutionInput) -> CodeExecutionResult:
         if not input_data.code_blocks and input_data.code:
             input_data.code_blocks = [CodeBlock(code=input_data.code, language="python")]
 
         kept: list[CodeBlock] = []
         blocked_msgs: list[str] = []
         for block in input_data.code_blocks:
-            report = scan(self._ensure_policy(), block.code,
-                          language=block.language or "auto")
+            report = scan(self._ensure_policy(), block.code, language=block.language or "auto")
             decision = report.decision
-            block_allowed = (
-                decision == Decision.ALLOW
-                or (decision == Decision.NEEDS_REVIEW and not self.block_on_review)
-            )
+            block_allowed = (decision == Decision.ALLOW
+                             or (decision == Decision.NEEDS_REVIEW and not self.block_on_review))
             # Audit each scanned block (issue #90): tool name, decision, risk,
             # rule ids, duration, sanitized, intercepted.
             record_safety_decision(
@@ -77,8 +73,7 @@ async def execute_code(self,
                 kept.append(block)
             else:
                 ids = ",".join(sorted({f.rule_id for f in report.findings}))
-                blocked_msgs.append(
-                    f"TOOL_SAFETY_BLOCKED [{block.language}] {decision.name} ({ids})")
+                blocked_msgs.append(f"TOOL_SAFETY_BLOCKED [{block.language}] {decision.name} ({ids})")
 
         out_parts: list[str] = []
         err_parts: list[str] = []

trpc_agent_sdk/tools/safety/_policy.py

@@ -78,9 +78,17 @@ def _policy_from_dict(raw: dict[str, Any]) -> Policy:
 
 
 _ALLOWED_TOP_LEVEL = {
-    "name", "description", "whitelisted_domains", "allowed_commands",
-    "denied_paths", "max_timeout_seconds", "max_output_bytes", "max_evidence_chars",
-    "deny_risk_level", "review_risk_level", "rule_overrides",
+    "name",
+    "description",
+    "whitelisted_domains",
+    "allowed_commands",
+    "denied_paths",
+    "max_timeout_seconds",
+    "max_output_bytes",
+    "max_evidence_chars",
+    "deny_risk_level",
+    "review_risk_level",
+    "rule_overrides",
 }
 
 
@@ -130,15 +138,17 @@ def _string_list(raw: dict[str, Any], key: str) -> list[str]:
 def _risk(raw: dict[str, Any], key: str, default: RiskLevel) -> RiskLevel:
     name = raw.get(key, default.name)
     if not isinstance(name, str):
-        raise ValueError(f"{key} must be a string enum name (one of {sorted(_VALID_RISK)}), got {type(name).__name__}: {name!r}")
+        raise ValueError(
+            f"{key} must be a string enum name (one of {sorted(_VALID_RISK)}), got {type(name).__name__}: {name!r}")
     if name not in _VALID_RISK:
         raise ValueError(f"{key} must be one of {sorted(_VALID_RISK)}, got {name!r}")
     return RiskLevel[name]
 
 
 def _decision(name: str) -> Decision:
     if not isinstance(name, str):
-        raise ValueError(f"decision must be a string enum name (one of {sorted(_VALID_DECISION)}), got {type(name).__name__}: {name!r}")
+        raise ValueError(f"decision must be a string enum name (one of {sorted(_VALID_DECISION)}), "
+                         f"got {type(name).__name__}: {name!r}")
     if name not in _VALID_DECISION:
         raise ValueError(f"decision must be one of {sorted(_VALID_DECISION)}, got {name!r}")
     return Decision[name]

trpc_agent_sdk/tools/safety/_python_scanner.py

@@ -24,17 +24,14 @@
 from trpc_agent_sdk.tools.safety._rules import R_RES_LARGE_WRITE
 from trpc_agent_sdk.tools.safety._rules import R_SECRET_LOGGING
 from trpc_agent_sdk.tools.safety._rules import R_SECRET_PRIVATE_KEY
-from trpc_agent_sdk.tools.safety._types import Decision
 from trpc_agent_sdk.tools.safety._types import Finding
-from trpc_agent_sdk.tools.safety._types import RiskLevel
 
 _CRED_PATH_RE = re.compile(r"(\.ssh|\.env|\.aws/credentials|id_rsa|id_ed25519|credentials)", re.I)
 _URL_RE = re.compile(r"https?://([^/\s'\"']+)", re.I)
 _SECRET_NAME_RE = re.compile(r"(api[_-]?key|secret|token|password|passwd|private[_-]?key)", re.I)
 _PRIVATE_KEY_RE = re.compile(r"-----BEGIN [A-Z ]*PRIVATE KEY-----")
 # System directories: writing here can brick the runtime (issue risk class #1).
-_SYSTEM_DIRS = ("/etc/", "/usr/", "/bin/", "/sbin/", "/boot/", "/sys/",
-                "/proc/", "/lib/", "/lib64/", "/var/", "/dev/")
+_SYSTEM_DIRS = ("/etc/", "/usr/", "/bin/", "/sbin/", "/boot/", "/sys/", "/proc/", "/lib/", "/lib64/", "/var/", "/dev/")
 
 # attribute path -> rule fired when called/used
 _DANGEROUS_ATTR = {
@@ -73,9 +70,13 @@ def scan_python(policy: Policy, script: str) -> list[Finding]:
 
     def add(rule_id: str, evidence: str, rec: str) -> None:
         meta = policy.rules[rule_id]
-        findings.append(Finding(
-            rule_id=rule_id, risk_level=meta.risk_level, rule_decision=meta.decision,
-            evidence=evidence[:max_ev], recommendation=rec, language="python"))
+        findings.append(
+            Finding(rule_id=rule_id,
+                    risk_level=meta.risk_level,
+                    rule_decision=meta.decision,
+                    evidence=evidence[:max_ev],
+                    recommendation=rec,
+                    language="python"))
 
     def resolve_attr(node: ast.AST) -> str:
         """Resolve `x.system` or `system` to 'module.attr' using alias tables."""
@@ -116,31 +117,28 @@ def resolve_attr(node: ast.AST) -> str:
                     add(R_NET_HTTP, f"{fname}({url})", f"{url} not whitelisted.")
             # Socket handled at module level to catch ALL socket.* calls (not a per-attribute list).
             if mod == "socket":
-                add(R_NET_SOCKET, f"{fname}()",
-                    "raw socket use bypasses HTTP allowlist; review egress.")
+                add(R_NET_SOCKET, f"{fname}()", "raw socket use bypasses HTTP allowlist; review egress.")
             # Resource abuse: very large writes and concurrency floods.
             if _is_write_call(fname) and _has_huge_size(node):
                 add(R_RES_LARGE_WRITE, f"{fname}(huge payload)",
                     "Very large write; possible disk/resource exhaustion. Review.")
             if _is_pool_call(fname) and _max_workers_too_large(node):
                 add(R_RES_CONCURRENT_FLOOD, f"{fname}(max_workers>>)",
                     "Very large worker pool; possible resource exhaustion. Review.")
-            if mod in ("open",) or fname.endswith(".open"):
+            if mod in ("open", ) or fname.endswith(".open"):
                 _check_open_path(node, policy, add)
         # infinite loop
-        if isinstance(node, (ast.While,)) and _is_truthy(node.test):
+        if isinstance(node, (ast.While, )) and _is_truthy(node.test):
             add(R_RES_INFINITE_LOOP, "while True:", "infinite loop; review.")
         # secret logging: assignment to a secret-named variable.
         if isinstance(node, ast.Assign):
             for t in node.targets:
                 if isinstance(t, ast.Name) and _SECRET_NAME_RE.search(t.id):
-                    add(R_SECRET_LOGGING, f"secret assigned to {t.id}",
-                        "secret-like variable; avoid logging.")
+                    add(R_SECRET_LOGGING, f"secret assigned to {t.id}", "secret-like variable; avoid logging.")
         # private-key literal embedded anywhere (independent of variable name).
         if isinstance(node, ast.Constant) and isinstance(node.value, str) \
                 and _PRIVATE_KEY_RE.search(node.value):
-            add(R_SECRET_PRIVATE_KEY, "private key literal",
-                "embedded private key; refuse.")
+            add(R_SECRET_PRIVATE_KEY, "private key literal", "embedded private key; refuse.")
 
     return findings
 
@@ -169,23 +167,19 @@ def _check_open_path(call: ast.Call, policy: Policy, add) -> None:
         path = "".join(v.value for v in arg.values if isinstance(v, ast.Constant))
     mode = _open_mode(call)
     if path and _is_write_mode(mode) and path.startswith(_SYSTEM_DIRS):
-        add(R_FS_SYSTEM_DIR, f"open('{path}','{mode}')",
-            f"writing to system directory {path}; refuse.")
+        add(R_FS_SYSTEM_DIR, f"open('{path}','{mode}')", f"writing to system directory {path}; refuse.")
         return
     if path and _CRED_PATH_RE.search(path):
-        add(R_FS_READ_CREDENTIALS, f"open('{path}')",
-            f"reading credential path {path}; review.")
+        add(R_FS_READ_CREDENTIALS, f"open('{path}')", f"reading credential path {path}; review.")
         return
     for denied in policy.denied_paths:
-        norm = denied.replace("~", "/root")  # crude home expansion for matching
         if denied in path:
-            add(R_FS_READ_CREDENTIALS, f"open('{path}')",
-                f"path matches denied path {denied}.")
+            add(R_FS_READ_CREDENTIALS, f"open('{path}')", f"path matches denied path {denied}.")
             return
 
 
 _HUGE_BYTES = 10_000_000  # ~10 MB heuristic threshold for "large write"
-_MAX_WORKERS = 100        # heuristic threshold for concurrency flood
+_MAX_WORKERS = 100  # heuristic threshold for concurrency flood
 
 _WRITE_METHODS = ("write", "write_bytes", "write_text")
 _POOL_CLASSES = ("ThreadPoolExecutor", "ProcessPoolExecutor")
@@ -209,9 +203,7 @@ def _huge_value(node: ast.AST) -> bool:
 
 
 def _has_huge_size(call: ast.Call) -> bool:
-    return any(_huge_value(a) for a in call.args) or any(
-        _huge_value(kw.value) for kw in call.keywords
-    )
+    return any(_huge_value(a) for a in call.args) or any(_huge_value(kw.value) for kw in call.keywords)
 
 
 def _max_workers_too_large(call: ast.Call) -> bool:
@@ -254,9 +246,13 @@ def _heuristic_fallback(policy: Policy, script: str) -> list[Finding]:
 
     def add(rule_id: str, evidence: str, rec: str) -> None:
         meta = policy.rules[rule_id]
-        findings.append(Finding(
-            rule_id=rule_id, risk_level=meta.risk_level, rule_decision=meta.decision,
-            evidence=evidence[:max_ev], recommendation=rec, language="python"))
+        findings.append(
+            Finding(rule_id=rule_id,
+                    risk_level=meta.risk_level,
+                    rule_decision=meta.decision,
+                    evidence=evidence[:max_ev],
+                    recommendation=rec,
+                    language="python"))
 
     if re.search(r"\beval\s*\(", script):
         add(R_CODE_UNSAFE_EVAL, "eval(", "eval executes arbitrary code.")

trpc_agent_sdk/tools/safety/_safety_filter.py

@@ -69,14 +69,11 @@ async def run(self, ctx: AgentContext, req: Any, handle: FilterHandleType) -> Fi
 
         script, language = extracted
         tool_name = getattr(ctx, "tool_name", None) or "tool_safety_filter"
-        report = scan(self._ensure_policy(), script, language=language,
-                      meta={"tool_name": tool_name})
+        report = scan(self._ensure_policy(), script, language=language, meta={"tool_name": tool_name})
         # Audit every decision (issue #90): allowed scripts get a summary,
         # denied scripts get a reason + an intercepted flag.
         intercepted = report.decision != Decision.ALLOW
-        record_safety_decision(
-            report, tool_name=tool_name, language=language, intercepted=intercepted
-        )
+        record_safety_decision(report, tool_name=tool_name, language=language, intercepted=intercepted)
         if report.decision == Decision.ALLOW:
             return await handle()
 

trpc_agent_sdk/tools/safety/_scanner.py

@@ -25,10 +25,7 @@ def detect_language(script: str) -> str:
     return "python"  # default
 
 
-def scan(policy: Policy,
-         script: str,
-         language: str = "auto",
-         meta: Optional[dict] = None) -> SafetyReport:
+def scan(policy: Policy, script: str, language: str = "auto", meta: Optional[dict] = None) -> SafetyReport:
     """Scan one script; return an aggregated SafetyReport.
 
     Args:
@@ -46,4 +43,4 @@ def scan(policy: Policy,
     elapsed_ms = int((time.perf_counter() - start) * 1000)
     report = aggregate(findings, policy)
     report.scan_duration_ms = elapsed_ms
-    return report
+    return report