fix(safety): apply same multi-block aggregation fix to UnsafeLocalCodeExecutor

AI Code Review critical pointed out TWO copies of the worst-report aggregation logic: _wrapper._scan_code_input (fixed in 1f15282) AND _unsafe_local_code_executor.py:130-149 (still had the bug). The executor copy also only promoted a block to worst on DENY or same-decision-higher-risk, so [safe_python (ALLOW), bash 'sleep 100 &' (NEEDS_HUMAN_REVIEW)] kept worst=ALLOW — same fail-open bypass. Apply the same (decision_rank, risk_rank) tuple comparison here.

Also fix AI Code Review suggestion: executor returned stderr='TOOL_SAFETY_DENY: ...' for BOTH deny and review-blocks, inconsistent with ToolSafetyFilter which distinguishes TOOL_SAFETY_DENY vs TOOL_SAFETY_NEEDS_REVIEW. Now picks the label based on report.decision so audits/logs don't mislabel review-blocks as deny.

Also document in safety_wrapper docstring that only keyword args are scanned (AI Code Review suggestion: positional args bypass _extract_script; documenting this is the lightweight fix since signature inspection is brittle for variadic callables).

Add two regression tests (run on CI where docker is installed): test_unsafe_local_code_executor_multi_block_aggregates_review and test_unsafe_local_code_executor_review_label_differs_from_deny.

Author: Violet2314

tests/tool_safety/test_opt_in.py

@@ -151,3 +151,43 @@ def test_unsafe_local_code_executor_safety_fields():
     assert ex._safety_scanner is not None
     result = asyncio.run(ex.execute_code(None, CodeExecutionInput(code="import os\nos.system('id')")))
     assert "TOOL_SAFETY_DENY" in result.output
+
+
+def test_unsafe_local_code_executor_multi_block_aggregates_review():
+    """Regression for AI Code Review critical: multi-block [safe_python, bash
+    'sleep 100 &'] must NOT keep worst=ALLOW. With block_on_review=True the
+    executor must block and emit TOOL_SAFETY_NEEDS_REVIEW."""
+    try:
+        from trpc_agent_sdk.code_executors.local._unsafe_local_code_executor import (
+            UnsafeLocalCodeExecutor, )
+        from trpc_agent_sdk.code_executors import CodeExecutionInput
+        from trpc_agent_sdk.code_executors import CodeBlock
+    except Exception as ex:  # pylint: disable=broad-except
+        pytest.skip(f"code executor not importable: {ex}")
+
+    ex = UnsafeLocalCodeExecutor(enable_safety_guard=True, safety_block_on_review=True)
+    inp = CodeExecutionInput(code_blocks=[
+        CodeBlock(language="python", code="x = 1\nprint(x)"),
+        CodeBlock(language="bash", code="sleep 100 &"),
+    ])
+    result = asyncio.run(ex.execute_code(None, inp))
+    # MUST be blocked (not run) and labeled as NEEDS_REVIEW, not DENY.
+    assert "TOOL_SAFETY_NEEDS_REVIEW" in result.output
+    assert "TOOL_SAFETY_DENY" not in result.output
+
+
+def test_unsafe_local_code_executor_review_label_differs_from_deny():
+    """When block_on_review=True and decision is NEEDS_HUMAN_REVIEW, stderr
+    must say TOOL_SAFETY_NEEDS_REVIEW, not TOOL_SAFETY_DENY (consistency with
+    ToolSafetyFilter). Single-block 'sleep 100 &' triggers MEDIUM/review."""
+    try:
+        from trpc_agent_sdk.code_executors.local._unsafe_local_code_executor import (
+            UnsafeLocalCodeExecutor, )
+        from trpc_agent_sdk.code_executors import CodeExecutionInput
+    except Exception as ex:  # pylint: disable=broad-except
+        pytest.skip(f"code executor not importable: {ex}")
+
+    ex = UnsafeLocalCodeExecutor(enable_safety_guard=True, safety_block_on_review=True)
+    result = asyncio.run(ex.execute_code(None, CodeExecutionInput(code="sleep 100 &")))
+    assert "TOOL_SAFETY_NEEDS_REVIEW" in result.output
+    assert "TOOL_SAFETY_DENY" not in result.output

trpc_agent_sdk/code_executors/local/_unsafe_local_code_executor.py

@@ -121,6 +121,14 @@ async def execute_code(self, invocation_context: InvocationContext,
                 RiskLevel.HIGH: 3,
                 RiskLevel.CRITICAL: 4,
             }
+            # Decision severity: DENY > NEEDS_HUMAN_REVIEW > ALLOW. A MEDIUM bash
+            # block must outweigh a safe ALLOW python block when aggregating
+            # multi-block input (matches _wrapper._scan_code_input).
+            _DECISION_RANK = {
+                Decision.ALLOW: 0,
+                Decision.NEEDS_HUMAN_REVIEW: 1,
+                Decision.DENY: 2,
+            }
             code_blocks = list(input_data.code_blocks or [])
             if not code_blocks and input_data.code:
                 # Prefer content inference over hardcoding python.
@@ -142,11 +150,13 @@ async def execute_code(self, invocation_context: InvocationContext,
                 report = self._safety_scanner.scan(ScanInput(script=code, language=lang, tool_name="code_executor"))
                 if worst is None:
                     worst = report
-                elif report.decision == Decision.DENY and worst.decision != Decision.DENY:
+                    continue
+                # Pick the worse of (worst, report) by (decision_rank, risk_rank)
+                # so NEEDS_HUMAN_REVIEW outweighs ALLOW even when risks differ.
+                worst_key = (_DECISION_RANK.get(worst.decision, 0), _ORDER.get(worst.risk_level, 0))
+                report_key = (_DECISION_RANK.get(report.decision, 0), _ORDER.get(report.risk_level, 0))
+                if report_key > worst_key:
                     worst = report
-                elif report.decision == worst.decision:
-                    if _ORDER.get(report.risk_level, 0) > _ORDER.get(worst.risk_level, 0):
-                        worst = report
 
             report = worst
             should_block = False
@@ -156,7 +166,11 @@ async def execute_code(self, invocation_context: InvocationContext,
                 if self._safety_audit is not None:
                     self._safety_audit.log(report, intercepted=should_block)
             if should_block and report is not None:
-                return create_code_execution_result(stderr=f"TOOL_SAFETY_DENY: {report.rule_ids}")
+                # Distinguish DENY vs NEEDS_HUMAN_REVIEW in stderr so audits and
+                # logs don't mislabel a review-block as a deny (matches
+                # ToolSafetyFilter's error_code selection).
+                code_label = ("TOOL_SAFETY_DENY" if report.decision == Decision.DENY else "TOOL_SAFETY_NEEDS_REVIEW")
+                return create_code_execution_result(stderr=f"{code_label}: {report.rule_ids}")
 
         output_parts = []
         error_parts = []

trpc_agent_sdk/safety/_wrapper.py

@@ -198,7 +198,17 @@ def safety_wrapper(
     audit_path=None,
     raise_on_deny=True,
 ):
-    """Decorator: scan the *script_arg* of a function before it runs."""
+    """Decorator: scan the *script_arg* of a function before it runs.
+
+    .. note::
+        Only keyword arguments are scanned. Positional arguments are not
+        mapped to *script_arg* (doing so reliably would require inspecting
+        the wrapped function's signature, which is brittle for *args/**kwargs
+        variadic callables). Callers MUST pass the script as a keyword
+        argument, e.g. ``run(script="rm -rf /")`` rather than
+        ``run("rm -rf /")``. A positional argument that is itself a dict
+        containing *script_arg* is also accepted as a legacy convenience.
+    """
     if policy is None:
         policy = PolicyConfig()
     _scanner = SafetyScanner(policy=policy)