fix: safety_wrapper fail-closed by default (require_script=True)

Author: lll-peanut

tests/test_tool_safety.py

@@ -1873,16 +1873,27 @@ def sync_tool(*args, **kwargs):
 
 
 def test_safety_wrapper_decorator_sync_no_script():
-    """@safety_wrapper sync must skip scan when script is None."""
+    """@safety_wrapper sync with require_script=False must skip scan and execute."""
 
-    @safety_wrapper(tool_name="sync_noscript_test", script_arg_name="code")
+    @safety_wrapper(tool_name="sync_noscript_test", script_arg_name="code", require_script=False)
     def sync_tool(*args, **kwargs):
         return "executed"
 
     result = sync_tool(args={})
     assert result == "executed"
 
 
+def test_safety_wrapper_decorator_sync_fail_closed():
+    """@safety_wrapper sync must raise RuntimeError when script arg is missing (fail-closed)."""
+
+    @safety_wrapper(tool_name="sync_failclosed_test", script_arg_name="code")
+    def sync_tool(*args, **kwargs):
+        return "executed"
+
+    with pytest.raises(RuntimeError, match="not found"):
+        sync_tool(args={})
+
+
 # ==========================================================================
 # Coverage gap tests — _telemetry.py
 # ==========================================================================
@@ -1975,16 +1986,27 @@ def test_dependency_install_python():
 
 
 async def test_safety_wrapper_decorator_async_no_script():
-    """@safety_wrapper async must skip scan when script is None."""
+    """@safety_wrapper async with require_script=False must skip scan and execute."""
 
-    @safety_wrapper(tool_name="async_noscript_test", script_arg_name="code")
+    @safety_wrapper(tool_name="async_noscript_test", script_arg_name="code", require_script=False)
     async def async_tool(*args, **kwargs):
         return "executed"
 
     result = await async_tool(args={})
     assert result == "executed"
 
 
+async def test_safety_wrapper_decorator_async_fail_closed():
+    """@safety_wrapper async must raise RuntimeError when script arg is missing (fail-closed)."""
+
+    @safety_wrapper(tool_name="async_failclosed_test", script_arg_name="code")
+    async def async_tool(*args, **kwargs):
+        return "executed"
+
+    with pytest.raises(RuntimeError, match="not found"):
+        await async_tool(args={})
+
+
 def test_scanner_empty_script():
     """Scanner must handle empty script content without crashing."""
     scanner = SafetyScanner()

trpc_agent_sdk/tools/safety/_safety_wrapper.py

@@ -189,6 +189,7 @@ def safety_wrapper(
     policy: Optional[SafetyPolicy] = None,
     audit_log_path: Optional[str] = None,
     raise_on_deny: bool = True,
+    require_script: bool = True,
 ):
     """Decorator that applies safety checks before a function executes.
 
@@ -201,6 +202,10 @@ def safety_wrapper(
         policy: Optional policy override.
         audit_log_path: Path to JSONL audit file.
         raise_on_deny: Raise ``SafetyDeniedError`` on DENY.
+        require_script: If True (default), raise ``RuntimeError`` when
+            *script_arg_name* is not found — fail-closed so that a
+            misconfigured decorator does not silently allow execution
+            without scanning.
 
     Example::
 
@@ -222,14 +227,17 @@ def decorator(func: Callable) -> Callable:
         async def async_wrapper(*args: Any, **kwargs: Any) -> Any:
             script: Optional[str] = kwargs.get(script_arg_name)
             if script is None:
-                # Try to find it in positional args (e.g. tool_context, args)
                 for arg in args:
                     if isinstance(arg, dict) and script_arg_name in arg:
                         script = arg[script_arg_name]
                         break
             if script and isinstance(script, str):
                 wrapper_inst.check(script)
-            elif script is not None and not isinstance(script, str):
+            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 "
+                                   f"or set require_script=False to skip scanning.")
+            elif script is not None:
                 _logger.warning("safety_wrapper: '%s' value is not a string (type=%s) — scan skipped.", script_arg_name,
                                 type(script).__name__)
             else:
@@ -248,7 +256,11 @@ def sync_wrapper(*args: Any, **kwargs: Any) -> Any:
                         break
             if script and isinstance(script, str):
                 wrapper_inst.check(script)
-            elif script is not None and not isinstance(script, str):
+            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 "
+                                   f"or set require_script=False to skip scanning.")
+            elif script is not None:
                 _logger.warning("safety_wrapper: '%s' value is not a string (type=%s) — scan skipped.", script_arg_name,
                                 type(script).__name__)
             else: