fix(safety): address CongkeChen review - bash path boundary, audit cross-process note, retry side effect doc

CI fix:
- _wrapper._deny_code_result fallback stub: yapf requires a blank line
  between class declaration and first method (E301). Added blank lines
  after `class _Outcome:` and `class _CodeExecutionResult:`.

Warning:
- _rules.DangerousFilesRule._check_bash: bash side system-dir and
  forbidden-path matching used substring (`sd in line` / `fb in line`),
  inconsistent with the Python side's path-boundary matching. So
  `cat /etcetera/foo` was false-positive CRITICAL → DENY (matched /etc),
  and `cat my.envrc` was false-positive → DENY (matched .env). Now both
  reuse _matches_system_dir(line) and _matches_forbidden(line, policy)
  so bash and Python sides agree.
- _audit._AUDIT_LOCK: add cross-process note explaining the lock is
  process-local and long JSONL lines (>PIPE_BUF, ~4096 bytes on Linux)
  may interleave/truncate across worker processes. Documents the three
  mitigations (per-worker path, single writer, fcntl.flock).
- _unsafe_local_code_executor.execute_code: document the
  OUTCOME_FAILED → error_retry_attempts side effect. Safety blocks map
  to OUTCOME_FAILED (non-empty stderr), which _code_execution_processor
  treats as an execution error and may retry up to 2 times. Each retry
  is still intercepted (safe), but wastes a scan/LLM round-trip. Comment
  notes a future refinement could use a dedicated marker to distinguish
  "safety block" from "real execution error".

Tests:
- test_dangerous_files_bash_system_dir_boundary_not_substring: assert
  `cat /etcetera/foo` does NOT trigger R001 (bash side, mirrors the
  existing Python-side regression test).
- test_dangerous_files_bash_forbidden_path_boundary_not_substring:
  assert `cat my.envrc` does NOT trigger forbidden-path R001 with
  policy forbidden_paths=[".env"].
- test_resource_long_sleep_above_threshold_is_high: assert sleep 301
  (max=300) is flagged HIGH.
- test_resource_sleep_at_threshold_is_allowed_bash: assert sleep 300
  (== max) is NOT flagged (strict > boundary).

Author: Violet2314

tests/tool_safety/test_rules.py

@@ -279,3 +279,49 @@ def test_strict_command_allowlist_empty_list_is_fail_closed():
     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)"
+
+
+def test_dangerous_files_bash_system_dir_boundary_not_substring():
+    """Regression for CongkeChen review: bash side used `sd in line` substring
+    matching, so '/etc' matched 'cat /etcetera/foo' (false-positive CRITICAL →
+    DENY). Now uses _matches_system_dir for path-boundary consistency with the
+    Python side."""
+    rule = DangerousFilesRule()
+    # '/etcetera/foo' must NOT match system dir '/etc' (no path boundary).
+    inp = ScanInput(script="cat /etcetera/foo\n", language="bash")
+    findings = rule.check(inp, _policy())
+    assert not any("system directory" in f.metadata.get("message", "") for f in findings)
+
+
+def test_dangerous_files_bash_forbidden_path_boundary_not_substring():
+    """Regression for CongkeChen review: bash side used `fb in line` substring
+    matching, so '.env' matched 'cat my.envrc' (false-positive → DENY). Now
+    uses _matches_forbidden for path-boundary consistency with the Python
+    side."""
+    rule = DangerousFilesRule()
+    policy = PolicyConfig(forbidden_paths=[".env"])
+    # 'my.envrc' must NOT match forbidden path '.env' (no path boundary).
+    inp = ScanInput(script="cat my.envrc\n", language="bash")
+    findings = rule.check(inp, policy)
+    assert not any("forbidden" in f.metadata.get("message", "").lower() for f in findings)
+
+
+def test_resource_long_sleep_above_threshold_is_high():
+    """sleep N where N > max_timeout_seconds must be flagged HIGH.
+    Threshold is > (strict), so max+1 triggers and max does not."""
+    from trpc_agent_sdk.safety._rules import ResourceAbuseRule
+    rule = ResourceAbuseRule()
+    policy = PolicyConfig(max_timeout_seconds=300)
+    inp = ScanInput(script="sleep 301\n", language="bash")
+    findings = rule.check(inp, policy)
+    assert any("Long sleep" in f.metadata.get("message", "") for f in findings)
+
+
+def test_resource_sleep_at_threshold_is_allowed_bash():
+    """sleep N where N == max_timeout_seconds must NOT be flagged (boundary)."""
+    from trpc_agent_sdk.safety._rules import ResourceAbuseRule
+    rule = ResourceAbuseRule()
+    policy = PolicyConfig(max_timeout_seconds=300)
+    inp = ScanInput(script="sleep 300\n", language="bash")
+    findings = rule.check(inp, policy)
+    assert not any("Long sleep" in f.metadata.get("message", "") for f in findings)

trpc_agent_sdk/code_executors/local/_unsafe_local_code_executor.py

@@ -127,6 +127,16 @@ async def execute_code(self, invocation_context: InvocationContext,
                 # 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).
+                #
+                # Side effect: create_code_execution_result maps non-empty
+                # stderr to OUTCOME_FAILED, and _code_execution_processor
+                # treats outcome != OUTCOME_OK as an execution error, which
+                # may trigger up to error_retry_attempts (default 2) retries
+                # of the same blocked code. Each retry is still intercepted
+                # (so this is safe), but it wastes a scan/LLM round-trip per
+                # retry. This is accepted for now; a future refinement could
+                # use a dedicated outcome/marker to distinguish "safety block"
+                # from "real execution error" so the agent does not retry.
                 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}")
             # Non-blocking NEEDS_HUMAN_REVIEW: do NOT block execution, but

trpc_agent_sdk/safety/_audit.py

@@ -16,6 +16,16 @@
 from ._types import SafetyReport
 
 # Process-local lock so concurrent threads do not interleave long JSONL lines.
+#
+# Cross-process note: this lock does NOT coordinate across OS processes. If
+# multiple worker processes write to the same audit_path simultaneously, long
+# JSONL lines (above PIPE_BUF, typically 4096 bytes on Linux) may interleave
+# or truncate because O_APPEND atomicity is only guaranteed up to PIPE_BUF.
+# For multi-process deployments, either (a) give each worker a distinct
+# audit_path, (b) funnel audit writes through a single writer process, or
+# (c) use fcntl.flock for cross-process exclusion. The current single-process
+# design matches the typical Tool/SafetyFilter usage where one agent process
+# owns the audit log.
 _AUDIT_LOCK = threading.Lock()
 
 

trpc_agent_sdk/safety/_rules.py

@@ -313,7 +313,11 @@ def _check_bash(self, scan_input: ScanInput, policy: PolicyConfig) -> list[Safet
                         ))
                     break
             for sd in _SYSTEM_DIRS:
-                if sd in line and (">" in line or "rm " in line or "chmod" in line or "chown" in line):
+                # Path-boundary match (consistent with _matches_system_dir on
+                # the Python side) so '/etc' matches '/etc/passwd' but NOT
+                # '/etcetera/foo'. Without this, safe scripts touching
+                # /etcetera would be false-positive CRITICAL → DENY.
+                if _matches_system_dir(line) and (">" in line or "rm " in line or "chmod" in line or "chown" in line):
                     findings.append(
                         self._finding(
                             line,
@@ -323,7 +327,11 @@ def _check_bash(self, scan_input: ScanInput, policy: PolicyConfig) -> list[Safet
                         ))
                     break
             for fb in policy.forbidden_paths:
-                if fb in line:
+                # Path-boundary match (consistent with _matches_forbidden on
+                # the Python side) so '.env' matches '/app/.env' but NOT
+                # 'my.envrc'. Without this, safe scripts touching my.envrc
+                # would be false-positive → DENY.
+                if _matches_forbidden(line, policy):
                     findings.append(
                         self._finding(
                             line,

trpc_agent_sdk/safety/_wrapper.py

@@ -68,10 +68,12 @@ def _deny_code_result(rule_ids: list[str]):
     )
 
     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)