fix(safety): multi-block aggregation must let NEEDS_HUMAN_REVIEW outweigh ALLOW

_scan_code_input's old aggregation only promoted a block to worst when it was DENY, or when it shared the same decision as worst with higher risk. So [safe_python (ALLOW), bash 'sleep 100 &' (NEEDS_HUMAN_REVIEW)] kept worst=ALLOW — a real bypass. Rewrite the comparison to rank by (decision_rank, risk_rank) where DENY > NEEDS_HUMAN_REVIEW > ALLOW, so any review/deny block outweighs an allow block.

Add two regression tests requested in review: (1) test_scan_code_input_multi_block_aggregates_safe_plus_review — multi-block [safe_python, 'sleep 100 &'] with block_on_review=True must return NEEDS_HUMAN_REVIEW and _should_block_report=True; (2) test_filter_block_on_review_pure_medium_returns_needs_review — ToolSafetyFilter with block_on_review=True and pure MEDIUM 'sleep 100 &' must return TOOL_SAFETY_NEEDS_REVIEW.

Author: Violet2314

tests/tool_safety/test_tool_filter.py

@@ -90,6 +90,23 @@ def test_filter_block_on_review(tmp_path: Path):
     assert rsp.rsp.get("success") is False
 
 
+def test_filter_block_on_review_pure_medium_returns_needs_review(tmp_path: Path):
+    """block_on_review=True + pure MEDIUM signal must return TOOL_SAFETY_NEEDS_REVIEW.
+
+    'sleep 100 &' is a non-network background process → MEDIUM →
+    NEEDS_HUMAN_REVIEW under the default policy (deny=HIGH, review=MEDIUM).
+    With block_on_review=True the filter must block and emit the review error
+    code (not DENY, since the decision is not DENY).
+    """
+    flt = _make_filter(tmp_path, block_on_review=True)
+    rsp = FilterResult()
+    req = {"command": "sleep 100 &"}
+    asyncio.run(flt._before(None, req, rsp))  # pylint: disable=protected-access
+    assert rsp.is_continue is False
+    assert rsp.rsp["error"] == "TOOL_SAFETY_NEEDS_REVIEW"
+    assert rsp.rsp.get("success") is False
+
+
 def test_filter_extracts_code_blocks(tmp_path: Path):
     flt = _make_filter(tmp_path)
     rsp = FilterResult()

tests/tool_safety/test_wrapper_language.py

@@ -93,3 +93,29 @@ def test_blocked_true_when_block_on_review_and_review_decision():
     report = SafetyScanner(policy).scan(ScanInput(script="sleep 1 &", language="bash"))
     assert report.decision.value == "needs_human_review"
     assert report.blocked is True
+
+
+def test_scan_code_input_multi_block_aggregates_safe_plus_review():
+    """Multi-block aggregation: safe Python + MEDIUM bash must yield NEEDS_HUMAN_REVIEW.
+
+    With block_on_review=True, _scan_code_input must return the worst report
+    (needs_human_review from the bash block, not allow from the safe python
+    block), and _should_block_report must return True.
+    """
+    from trpc_agent_sdk.safety._wrapper import _should_block_report
+    from trpc_agent_sdk.safety import Decision
+
+    policy = PolicyConfig(
+        deny_risk_level=RiskLevel.CRITICAL,
+        review_risk_level=RiskLevel.MEDIUM,
+        block_on_review=True,
+    )
+    scanner = SafetyScanner(policy)
+    inp = _Input(code_blocks=[
+        _Block("x = 1\nprint(x)", language="python"),
+        _Block("sleep 100 &", language="bash"),
+    ])
+    report = _scan_code_input(scanner, inp)
+    assert report is not None
+    assert report.decision == Decision.NEEDS_HUMAN_REVIEW
+    assert _should_block_report(report, block_on_review=True) is True

trpc_agent_sdk/safety/_wrapper.py

@@ -63,7 +63,14 @@ def _normalize_block_language(raw_lang: str | None, code: str) -> str:
 
 def _scan_code_input(scanner: SafetyScanner, input_data):
     """Scan code / code_blocks with per-block language; return worst report."""
-    order = {
+    # Decision severity: DENY > NEEDS_HUMAN_REVIEW > ALLOW. A MEDIUM bash block
+    # must outweigh a safe ALLOW python block when aggregating multi-block input.
+    decision_rank = {
+        Decision.ALLOW: 0,
+        Decision.NEEDS_HUMAN_REVIEW: 1,
+        Decision.DENY: 2,
+    }
+    risk_rank = {
         RiskLevel.NONE: 0,
         RiskLevel.LOW: 1,
         RiskLevel.MEDIUM: 2,
@@ -91,11 +98,12 @@ def _scan_code_input(scanner: SafetyScanner, input_data):
         report = 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).
+        worst_key = (decision_rank.get(worst.decision, 0), risk_rank.get(worst.risk_level, 0))
+        report_key = (decision_rank.get(report.decision, 0), risk_rank.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
     return worst