fix(safety): address CongkeChen review - real CodeExecutionResult, fail-closed allowlist, dict blocks, logging

CI fix:
- _rules._matches_forbidden: yapf wanted the multi-or condition on one line
  (it fits in 120 chars). Reformatted to match yapf output.

Critical:
- _wrapper._deny_code_result: replace custom _Outcome/_CodeExecutionResult
  stub objects with real trpc_agent_sdk.types.CodeExecutionResult +
  Outcome.OUTCOME_FAILED. The stub's _Outcome only had .name, so downstream
  pipeline code (Part.from_code_execution_result, _openai_model) that reads
  .outcome.value would AttributeError. trpc_agent_sdk.types pulls
  google.genai.types (always installed with SDK core), NOT docker, so the
  import stays dep-light. A fallback stub is kept only for the impossible
  case where the SDK core itself is broken, and logs an error so the
  misconfiguration is diagnosable.

Warning:
- _rules.ProcessRule._check_bash: strict_command_allowlist=True with empty
  allowed_commands used to skip the whole allow-list check (fail-open),
  letting rm/chmod slip through. Now every non-builtin command is flagged
  HIGH (fail-closed), matching the semantic of "strict allow-list".
- _wrapper._scan_code_input: support dict-form code_blocks (with "code"/
  "language" keys) in addition to object-form. Previously getattr on a
  dict returned None/empty, silently skipping real code in dict blocks.
- _unsafe_local_code_executor.execute_code: move _pending_review_warning
  initialization to the top of the function so it always has a defined
  value regardless of which branch runs. Previously it was only assigned
  inside the if/else branches, making later `if _pending_review_warning`
  depend on branch ordering (fragile for refactors).
- _filter._before + _unsafe_local_code_executor: replace sys.stderr.write
  with logging module (_logger.warning). Direct stderr writes pollute
  captured stderr streams in CI / subprocess stdout-stderr separation, and
  per-call stderr writes could be noisy. The audit log already records the
  decision; the log line is the in-band fallback for operators.

Tests:
- test_wrapper: assert against real Outcome.OUTCOME_FAILED /
  OUTCOME_OK (with .value access) so the test fails if a stub object is
  ever reintroduced. _FakeInnerExecutor now returns a real
  CodeExecutionResult too.
- test_rules: add test_strict_command_allowlist_empty_list_is_fail_closed
  to lock in the fail-closed behavior for empty allowed_commands.

Author: Violet2314

tests/tool_safety/test_rules.py

@@ -264,3 +264,18 @@ def test_dangerous_files_system_dir_boundary_not_substring():
     findings = rule.check(inp, _policy())
     # No system-dir finding should fire for /etcetera.
     assert not any("system" in f.metadata.get("message", "").lower() for f in findings)
+
+
+def test_strict_command_allowlist_empty_list_is_fail_closed():
+    """Regression for CongkeChen review: strict_command_allowlist=True with
+    empty allowed_commands used to skip the whole allow-list check (fail-open),
+    letting rm/chmod through. Now every non-builtin command is flagged HIGH."""
+    from trpc_agent_sdk.safety._rules import ProcessRule
+    # Use ProcessRule's _check_bash path (DangerousFilesRule also runs but
+    # ProcessRule carries the allow-list enforcement).
+    rule = ProcessRule()
+    policy = PolicyConfig(strict_command_allowlist=True, allowed_commands=[])
+    inp = ScanInput(script="rm /tmp/x\n", language="bash")
+    findings = rule.check(inp, policy)
+    allow_list_findings = [f for f in findings if "allow-list" in f.metadata.get("message", "")]
+    assert allow_list_findings, "strict_command_allowlist=True with empty allowed_commands must flag rm (fail-closed)"

tests/tool_safety/test_wrapper.py

@@ -26,6 +26,13 @@
 except Exception:  # pylint: disable=broad-except
     FilterResult = None  # type: ignore[assignment]
 
+try:
+    from trpc_agent_sdk.types import Outcome
+    _REAL_OUTCOME_AVAILABLE = True
+except Exception:  # pylint: disable=broad-except
+    Outcome = None  # type: ignore[assignment]
+    _REAL_OUTCOME_AVAILABLE = False
+
 pytestmark = pytest.mark.skipif(
     not _SDK_AVAILABLE or FilterResult is None,
     reason="tRPC-Agent SDK core (abc.FilterResult) not importable",
@@ -95,12 +102,12 @@ def __init__(self) -> None:
 
     async def execute_code(self, invocation_context, input_data):
         self.calls.append(input_data)
-
-        class _Ok:
-            output = "ok"
-            outcome = type("O", (), {"name": "OUTCOME_OK"})()
-
-        return _Ok()
+        # Return a real CodeExecutionResult so the test exercises the same
+        # types downstream pipeline code (Part.from_code_execution_result,
+        # _openai_model) will see, instead of a stub that hides attr errors.
+        from trpc_agent_sdk.types import CodeExecutionResult
+        from trpc_agent_sdk.types import Outcome
+        return CodeExecutionResult(outcome=Outcome.OUTCOME_OK, output="ok")
 
 
 def test_safe_code_executor_blocks_dangerous_python(tmp_path: Path):
@@ -113,7 +120,14 @@ def test_safe_code_executor_blocks_dangerous_python(tmp_path: Path):
 
     assert inner.calls == []
     assert "TOOL_SAFETY_DENY" in result.output
-    assert result.outcome.name == "OUTCOME_FAILED"
+    # Real Outcome enum: .value / .name both available. Assert against the
+    # enum member directly so the test fails if a stub object is ever
+    # reintroduced (stubs lack .value and would fail equality).
+    if _REAL_OUTCOME_AVAILABLE:
+        assert result.outcome == Outcome.OUTCOME_FAILED
+        assert result.outcome.value is not None  # stub would crash here
+    else:  # pragma: no cover - fallback when SDK types unavailable
+        assert result.outcome.name == "OUTCOME_FAILED"
 
     audit_path = tmp_path / "audit.jsonl"
     assert audit_path.exists()
@@ -132,7 +146,10 @@ def test_safe_code_executor_allows_safe_python(tmp_path: Path):
 
     assert len(inner.calls) == 1
     assert "ok" in result.output
-    assert result.outcome.name == "OUTCOME_OK"
+    if _REAL_OUTCOME_AVAILABLE:
+        assert result.outcome == Outcome.OUTCOME_OK
+    else:  # pragma: no cover - fallback when SDK types unavailable
+        assert result.outcome.name == "OUTCOME_OK"
 
 
 def test_safety_wrapper_blocks_dangerous_script(tmp_path: Path):

trpc_agent_sdk/code_executors/local/_unsafe_local_code_executor.py

@@ -105,6 +105,12 @@ async def execute_code(self, invocation_context: InvocationContext,
         Returns:
             CodeExecutionResult with combined output
         """
+        # Initialize at the top so the variable always has a defined value
+        # regardless of which branch (scanner on/off, report None/not) runs.
+        # Previously it was only assigned inside the `if self._safety_scanner`
+        # branch and the `else`, making later `if _pending_review_warning`
+        # depend on branch ordering — fragile for refactors.
+        _pending_review_warning = None
         if self._safety_scanner is not None:
             from trpc_agent_sdk.safety import Decision
             from trpc_agent_sdk.safety._wrapper import _scan_code_input
@@ -125,14 +131,13 @@ async def execute_code(self, invocation_context: InvocationContext,
                 return create_code_execution_result(stderr=f"{code_label}: {report.rule_ids}")
             # Non-blocking NEEDS_HUMAN_REVIEW: do NOT block execution, but
             # surface the warning so the caller is not silently unaware
-            # (matches _filter._after's merge semantics). We stash the warning
+            # (matches _filter._after's merge semantics). Stash the warning
             # and prepend it to stderr after execution completes below.
-            _pending_review_warning = None
+            # Use logging (not sys.stderr.write) to avoid polluting captured
+            # stderr streams in CI / subprocess stdout/stderr separation.
             if (report is not None and report.decision == Decision.NEEDS_HUMAN_REVIEW and not should_block):
                 _pending_review_warning = (f"TOOL_SAFETY_NEEDS_REVIEW: {list(report.rule_ids)} "
                                            f"(risk={report.risk_level.value})")
-        else:
-            _pending_review_warning = None
 
         output_parts = []
         error_parts = []

trpc_agent_sdk/safety/_filter.py

@@ -7,7 +7,7 @@
 from __future__ import annotations
 
 import contextvars
-import sys
+import logging
 from typing import Any
 from typing import Optional
 
@@ -21,6 +21,8 @@
 from ._types import Decision
 from ._types import ScanInput
 
+_logger = logging.getLogger("trpc_agent_sdk.safety")
+
 _SCRIPT_ARG_KEYS = ("command", "script", "code", "cmd", "bash", "shell_command")
 _LANGUAGE_ARG_KEYS = ("language", "lang")
 _WORKDIR_ARG_KEYS = ("cwd", "workdir", "working_dir")
@@ -130,14 +132,15 @@ async def _before(self, ctx: Any, req: Any, rsp: FilterResult) -> None:
             # BaseFilter.run's handle() step overwrites result.rsp with the
             # tool's return value.
             #
-            # Belt-and-suspenders: also write to stderr so that if _after is
-            # skipped (handle() raised, earlier filter set is_continue=False,
-            # or the tool returned a non-dict that _after cannot annotate),
-            # the warning is still surfaced to the caller instead of being
-            # silently lost. The audit log already records the decision; this
-            # stderr line is the in-band fallback for human operators.
-            sys.stderr.write(f"TOOL_SAFETY_NEEDS_REVIEW: {list(report.rule_ids)} "
-                             f"(risk={report.risk_level.value})\n")
+            # Belt-and-suspenders: also log via the SDK logger so that if
+            # _after is skipped (handle() raised, earlier filter set
+            # is_continue=False, or the tool returned a non-dict that _after
+            # cannot annotate), the warning is still surfaced to operators
+            # instead of being silently lost. The audit log already records
+            # the decision; this log line is the in-band fallback. Using the
+            # logging module (not sys.stderr.write) avoids polluting captured
+            # stderr streams in CI / subprocess scenarios.
+            _logger.warning("TOOL_SAFETY_NEEDS_REVIEW: %s (risk=%s)", list(report.rule_ids), report.risk_level.value)
             _REVIEW_REPORT.set(report)
 
     async def _after(self, ctx: Any, req: Any, rsp: FilterResult) -> None:

trpc_agent_sdk/safety/_rules.py

@@ -153,9 +153,7 @@ def _matches_forbidden(target: str, policy: PolicyConfig) -> bool:
         fb_norm = fb.replace("\\", "/")
         if not fb_norm:
             continue
-        if (tgt_norm == fb_norm
-                or tgt_norm.startswith(fb_norm + "/")
-                or tgt_norm.endswith("/" + fb_norm)
+        if (tgt_norm == fb_norm or tgt_norm.startswith(fb_norm + "/") or tgt_norm.endswith("/" + fb_norm)
                 or ("/" not in tgt_norm and tgt_norm == fb_norm)):
             return True
     return False
@@ -848,9 +846,12 @@ def _check_bash(self, scan_input: ScanInput, policy: PolicyConfig) -> list[Safet
                 continue
             cmd = tokens[0].split("/")[-1]
 
-            if policy.strict_command_allowlist and policy.allowed_commands:
-                # Skip env assignments; reuse PolicyConfig.is_command_allowed
-                # so Windows basenames (\) stay consistent with policy helpers.
+            if policy.strict_command_allowlist:
+                # Fail-closed: when strict mode is on but allowed_commands is
+                # empty, every non-builtin command is unknown and must be
+                # flagged HIGH. The previous `and policy.allowed_commands`
+                # guard turned strict mode into a no-op for empty allow-lists,
+                # letting rm/chmod slip through.
                 check_cmd = cmd
                 if "=" in check_cmd and len(tokens) > 1:
                     check_cmd = tokens[1].split("/")[-1].split("\\")[-1]

trpc_agent_sdk/safety/_wrapper.py

@@ -8,6 +8,7 @@
 
 import asyncio
 import functools
+import logging
 from typing import Optional
 
 from ._audit import AuditLogger
@@ -19,6 +20,23 @@
 from ._types import decision_rank
 from ._types import risk_order
 
+# Use the real CodeExecutionResult / Outcome from trpc_agent_sdk.types so deny
+# results flow through the Agent pipeline (Part.from_code_execution_result,
+# _openai_model._post_process_code_execution_result) without AttributeError on
+# .outcome.value. trpc_agent_sdk.types pulls google.genai.types (always
+# installed with the SDK core), NOT docker, so this stays dep-light.
+try:  # pragma: no cover
+    from trpc_agent_sdk.types import CodeExecutionResult
+    from trpc_agent_sdk.types import Outcome
+    _REAL_RESULT_TYPES_AVAILABLE = True
+except Exception as ex:  # pylint: disable=broad-except
+    CodeExecutionResult = None  # type: ignore[assignment]
+    Outcome = None  # type: ignore[assignment]
+    _REAL_RESULT_TYPES_AVAILABLE = False
+    _RESULT_IMPORT_ERROR = ex
+
+_logger = logging.getLogger("trpc_agent_sdk.safety")
+
 
 def wrap_tool(tool, policy: PolicyConfig, *, audit_path: Optional[str] = None):
     """Return *tool* with a :class:`ToolSafetyFilter` prepended to its filters."""
@@ -27,24 +45,37 @@ def wrap_tool(tool, policy: PolicyConfig, *, audit_path: Optional[str] = None):
     return tool
 
 
-class _Outcome:
-    """Minimal stand-in for trpc_agent_sdk.types.Outcome used in deny results."""
-
-    def __init__(self, name: str):
-        self.name = name
-
-
-class _CodeExecutionResult:
-    """Minimal CodeExecutionResult-compatible object (no code_executors import)."""
-
-    def __init__(self, *, output: str, outcome_name: str):
-        self.output = output
-        self.outcome = _Outcome(outcome_name)
-
+def _deny_code_result(rule_ids: list[str]):
+    """Build a failed CodeExecutionResult using the real SDK types.
 
-def _deny_code_result(rule_ids: list[str]) -> _CodeExecutionResult:
-    """Build a failed code-execution result without importing code_executors."""
+    Falls back to a minimal stand-in only if ``trpc_agent_sdk.types`` cannot be
+    imported (e.g. google-genai missing), which would itself break the whole
+    SDK. The real ``Outcome.OUTCOME_FAILED`` is an enum with ``.value`` and
+    ``.name``, so downstream code (Part.from_code_execution_result,
+    _openai_model) that reads ``.outcome.value`` works correctly.
+    """
     msg = f"Code execution error:\nTOOL_SAFETY_DENY: {rule_ids}\n"
+    if _REAL_RESULT_TYPES_AVAILABLE:
+        return CodeExecutionResult(outcome=Outcome.OUTCOME_FAILED, output=msg)
+    # Last-resort fallback (SDK core itself is broken). Emit a clear error so
+    # the misconfiguration is diagnosable instead of silently producing a
+    # stub object that crashes downstream pipeline code.
+    _logger.error(
+        "trpc_agent_sdk.types import failed; cannot build real "
+        "CodeExecutionResult. Deny result will use a minimal stub. "
+        "Original error: %r",
+        _RESULT_IMPORT_ERROR,
+    )
+
+    class _Outcome:
+        def __init__(self, name: str):
+            self.name = name
+
+    class _CodeExecutionResult:
+        def __init__(self, *, output: str, outcome_name: str):
+            self.output = output
+            self.outcome = _Outcome(outcome_name)
+
     return _CodeExecutionResult(output=msg, outcome_name="OUTCOME_FAILED")
 
 
@@ -63,20 +94,37 @@ 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."""
+    """Scan code / code_blocks with per-block language; return worst report.
+
+    ``code_blocks`` elements may be either objects (with ``.code`` /
+    ``.language`` attributes, e.g. ``CodeBlock``) or dicts (with ``code`` /
+    ``language`` keys, as produced by some tool adapters). Both forms are
+    supported so the scanner does not silently skip dict blocks (which would
+    leave real code unscanned).
+    """
     # Use shared ranking from _types so this aggregation logic cannot drift
     # from the executor's copy. Decision severity: DENY > NEEDS_HUMAN_REVIEW >
     # ALLOW. A MEDIUM bash block must outweigh a safe ALLOW python block.
     blocks = list(getattr(input_data, "code_blocks", None) or [])
     top_code = getattr(input_data, "code", None) or ""
     if not blocks and top_code:
         top_lang = getattr(input_data, "language", None) or ""
-        blocks = [type("B", (), {"code": top_code, "language": top_lang})()]
+        blocks = [{"code": top_code, "language": top_lang}]
+
+    def _block_code(block) -> str:
+        if isinstance(block, dict):
+            return str(block.get("code", "") or "")
+        return str(getattr(block, "code", None) or "")
+
+    def _block_lang(block) -> str:
+        if isinstance(block, dict):
+            return str(block.get("language", "") or "").strip().lower()
+        return str(getattr(block, "language", None) or "").strip().lower()
 
     worst = None
     for block in blocks:
-        code = getattr(block, "code", None) or ""
-        declared = (getattr(block, "language", None) or "").strip().lower()
+        code = _block_code(block)
+        declared = _block_lang(block)
         # Mislabeled bash-as-python must still hit bash rules via content check.
         if declared in ("sh", "shell", "bash"):
             lang = "bash"