fix: pass command_args/env to decorator and filter, W504 line break

  - _safety_wrapper: decorator now passes command_args/environment_variables/
    working_directory to SafetyWrapper.check (was dead code path)
  - _safety_filter: _before now extracts command_args/env_vars/work_dir from req
  - _scanner: fix W504 line break after binary operator

Author: lll-peanut

trpc_agent_sdk/tools/safety/_safety_filter.py

@@ -113,10 +113,17 @@ async def _before(self, ctx: AgentContext, req: Any, rsp: FilterResult) -> None:
         script_type = _guess_script_type(req, script_content)
         tool_name = _extract_tool_name(req)
 
+        cmd_args = _extract_list_field(req, "command_args", "cmd_args")
+        env_vars = _extract_dict_field(req, "environment_variables", "env_vars")
+        work_dir = _extract_str_field(req, "working_directory", "cwd")
+
         scan_input = SafetyScanInput(
             script_content=script_content,
             script_type=script_type,
             tool_name=tool_name,
+            command_args=cmd_args,
+            environment_variables=env_vars,
+            working_directory=work_dir,
         )
 
         report = self._scanner.scan(scan_input)
@@ -239,3 +246,40 @@ def _extract_tool_name(req: Any) -> str:
     if hasattr(req, "name"):
         return str(getattr(req, "name", "unknown"))
     return "unknown"
+
+
+def _extract_list_field(req: Any, *keys: str) -> Optional[list[str]]:
+    """Extract a list-typed field from the request by trying multiple key names."""
+    if isinstance(req, dict):
+        for k in keys:
+            val = req.get(k) or req.get("args", {}).get(k)
+            if isinstance(val, list):
+                return val
+    if hasattr(req, "args"):
+        args = getattr(req, "args")
+        if isinstance(args, dict):
+            for k in keys:
+                val = args.get(k)
+                if isinstance(val, list):
+                    return val
+    return None
+
+
+def _extract_str_field(req: Any, *keys: str) -> Optional[str]:
+    """Extract a string-typed field from the request by trying multiple key names."""
+    if isinstance(req, dict):
+        for k in keys:
+            val = req.get(k) or req.get("args", {}).get(k)
+            if isinstance(val, str):
+                return val
+    return None
+
+
+def _extract_dict_field(req: Any, *keys: str) -> Optional[dict[str, str]]:
+    """Extract a dict-typed field from the request by trying multiple key names."""
+    if isinstance(req, dict):
+        for k in keys:
+            val = req.get(k) or req.get("args", {}).get(k)
+            if isinstance(val, dict):
+                return val
+    return None

trpc_agent_sdk/tools/safety/_safety_wrapper.py

@@ -223,6 +223,16 @@ def decorator(func: Callable) -> Callable:
             raise_on_deny=raise_on_deny,
         )
 
+        def _extract_extra_fields(call_args: tuple, call_kwargs: dict) -> tuple:
+            """Extract extra scan fields from args/kwargs."""
+            for arg in call_args:
+                if isinstance(arg, dict):
+                    return (arg.get("command_args") or arg.get("cmd_args"), arg.get("environment_variables")
+                            or arg.get("env_vars"), arg.get("working_directory") or arg.get("cwd"))
+            return (call_kwargs.get("command_args")
+                    or call_kwargs.get("cmd_args"), call_kwargs.get("environment_variables")
+                    or call_kwargs.get("env_vars"), call_kwargs.get("working_directory") or call_kwargs.get("cwd"))
+
         @functools.wraps(func)
         async def async_wrapper(*args: Any, **kwargs: Any) -> Any:
             script: Optional[str] = kwargs.get(script_arg_name)
@@ -232,7 +242,11 @@ async def async_wrapper(*args: Any, **kwargs: Any) -> Any:
                         script = arg[script_arg_name]
                         break
             if script and isinstance(script, str):
-                wrapper_inst.check(script)
+                cmd_args, env_vars, work_dir = _extract_extra_fields(args, kwargs)
+                wrapper_inst.check(script,
+                                   command_args=cmd_args,
+                                   environment_variables=env_vars,
+                                   working_directory=work_dir)
             elif require_script:
                 raise RuntimeError(f"safety_wrapper: '{script_arg_name}' not found in kwargs or positional "
                                    f"dict args for {func.__name__}.  Check the decorator configuration "
@@ -255,7 +269,11 @@ def sync_wrapper(*args: Any, **kwargs: Any) -> Any:
                         script = arg[script_arg_name]
                         break
             if script and isinstance(script, str):
-                wrapper_inst.check(script)
+                cmd_args, env_vars, work_dir = _extract_extra_fields(args, kwargs)
+                wrapper_inst.check(script,
+                                   command_args=cmd_args,
+                                   environment_variables=env_vars,
+                                   working_directory=work_dir)
             elif require_script:
                 raise RuntimeError(f"safety_wrapper: '{script_arg_name}' not found in kwargs or positional "
                                    f"dict args for {func.__name__}.  Check the decorator configuration "

trpc_agent_sdk/tools/safety/_scanner.py

@@ -281,9 +281,9 @@ def scan(self, scan_input: SafetyScanInput) -> SafetyScanReport:
         else:
             denied = sum(1 for f in all_findings if f.risk_level in (RiskLevel.CRITICAL, RiskLevel.HIGH))
             total = len(all_findings)
+            auto_suffix = " [auto_allowed by allow_patterns]" if allow_upgraded else ""
             summary = (f"Scan of '{scan_input.tool_name or 'unnamed tool'}' found {total} issue(s) "
-                       f"({denied} high/critical). Decision: {decision.value}." +
-                       (" [auto_allowed by allow_patterns]" if allow_upgraded else ""))
+                       f"({denied} high/critical). Decision: {decision.value}.{auto_suffix}")
 
         return SafetyScanReport(
             tool_name=scan_input.tool_name,