fix(safety): 提交测序策略文件，修复代码规范问题

Author: ha0924

tests/tools/safety/fixtures/sample_policy.yaml

@@ -0,0 +1,26 @@
+# Sample policy for integration testing
+# This adds extra allowed domains and commands beyond the built-in defaults.
+version: "1.0"
+
+network:
+  allowed_domains:
+    - "*.example.com"
+    - "custom-api.mycompany.io"
+  override: false  # Append to defaults
+
+process:
+  allowed_commands:
+    - "docker"
+    - "go"
+    - "cargo"
+  override: false
+
+file_operations:
+  forbidden_paths:
+    - "/tmp/sensitive/"
+    - "~/secrets/"
+  override: false
+
+resources:
+  max_timeout_seconds: 600
+  max_output_size_mb: 200

tests/tools/safety/fixtures/strict_policy.yaml

@@ -0,0 +1,30 @@
+# Strict policy for integration testing — very restrictive
+# Override mode: fully replace default lists with minimal allowlists
+version: "1.0"
+
+network:
+  allowed_domains:
+    - "pypi.org"
+  override: true  # Only pypi.org allowed, all others blocked
+
+process:
+  allowed_commands:
+    - "echo"
+    - "cat"
+  override: true  # Only echo and cat allowed
+
+file_operations:
+  forbidden_paths:
+    - "/etc/"
+    - "/root/"
+    - "~/.ssh/"
+    - "~/.aws/"
+    - "~/.gnupg/"
+    - "~/.config/"
+    - "/var/"
+    - "/tmp/"
+  override: true  # Strictest set
+
+resources:
+  max_timeout_seconds: 60
+  max_output_size_mb: 10

tests/tools/safety/test_guard.py

@@ -1,7 +1,6 @@
 """Unit tests for ScriptSafetyGuard — the core coordination engine."""
 
 import json
-import logging
 from unittest.mock import patch
 
 import pytest
@@ -257,19 +256,16 @@ def test_multiple_deny_still_deny(self):
 class TestAuditLog:
     """Test that audit logging is emitted correctly."""
 
-    def test_audit_log_emitted_on_check(self, caplog):
+    def test_audit_log_emitted_on_check(self):
         """Verify audit log is produced with expected structure."""
         guard = ScriptSafetyGuard()
         input = _make_input(script="x = 1")
 
-        with caplog.at_level(logging.INFO, logger="trpc_agent_sdk.tools.safety.audit"):
+        with patch("trpc_agent_sdk.tools.safety.guard._audit_logger") as mock_audit:
             guard.check(input)
 
-        # Find the audit log record
-        audit_records = [r for r in caplog.records if r.name == "trpc_agent_sdk.tools.safety.audit"]
-        assert len(audit_records) == 1
-
-        entry = json.loads(audit_records[0].message)
+        mock_audit.info.assert_called_once()
+        entry = json.loads(mock_audit.info.call_args[0][0])
         assert entry["event"] == "safety_check"
         assert entry["decision"] == "allow"
         assert entry["language"] == "python"
@@ -279,18 +275,16 @@ def test_audit_log_emitted_on_check(self, caplog):
         assert "findings_count" in entry
         assert "script_length" in entry
 
-    def test_audit_log_contains_findings_summary(self, caplog):
+    def test_audit_log_contains_findings_summary(self):
         """Audit log findings are present with desensitized evidence."""
         guard = ScriptSafetyGuard()
         input = _make_input(script="import os\nos.system('rm -rf /')")
 
-        with caplog.at_level(logging.INFO, logger="trpc_agent_sdk.tools.safety.audit"):
+        with patch("trpc_agent_sdk.tools.safety.guard._audit_logger") as mock_audit:
             guard.check(input)
 
-        audit_records = [r for r in caplog.records if r.name == "trpc_agent_sdk.tools.safety.audit"]
-        assert len(audit_records) == 1
-
-        entry = json.loads(audit_records[0].message)
+        mock_audit.info.assert_called_once()
+        entry = json.loads(mock_audit.info.call_args[0][0])
         assert entry["findings_count"] > 0
         assert len(entry["findings"]) > 0
         # Each finding has expected fields

tests/tools/safety/test_guard_extended.py

@@ -206,10 +206,8 @@ def test_audit_write_failure_does_not_raise(self):
 class TestEmitAuditLogExtended:
     """Extended tests for the Python logger audit log."""
 
-    def test_audit_log_with_findings(self, caplog):
+    def test_audit_log_with_findings(self):
         """Audit log includes sanitized findings."""
-        import logging
-
         input_data = _make_input(script="api_key = 'sk-secret1234567890abcdefgh'")
         finding = Finding(
             rule_id="SEC-001",
@@ -229,31 +227,28 @@ def test_audit_log_with_findings(self, caplog):
             invocation_id="inv-x",
         )
 
-        with caplog.at_level(logging.INFO, logger="trpc_agent_sdk.tools.safety.audit"):
+        with patch("trpc_agent_sdk.tools.safety.guard._audit_logger") as mock_audit:
             _emit_audit_log(input_data, result)
 
-        audit_records = [r for r in caplog.records if r.name == "trpc_agent_sdk.tools.safety.audit"]
-        assert len(audit_records) == 1
-        entry = json.loads(audit_records[0].message)
+        mock_audit.info.assert_called_once()
+        entry = json.loads(mock_audit.info.call_args[0][0])
         assert entry["decision"] == "deny"
         assert entry["findings_count"] == 1
         # Evidence should be sanitized in audit log
         f = entry["findings"][0]
         assert "****" in f["evidence"]
         assert f["line_number"] == 1
 
-    def test_audit_log_metadata_fields(self, caplog):
+    def test_audit_log_metadata_fields(self):
         """Audit log contains agent_name, user_id, script_length."""
-        import logging
-
         input_data = _make_input(script="x = 42")
         result = _make_result()
 
-        with caplog.at_level(logging.INFO, logger="trpc_agent_sdk.tools.safety.audit"):
+        with patch("trpc_agent_sdk.tools.safety.guard._audit_logger") as mock_audit:
             _emit_audit_log(input_data, result)
 
-        audit_records = [r for r in caplog.records if r.name == "trpc_agent_sdk.tools.safety.audit"]
-        entry = json.loads(audit_records[0].message)
+        mock_audit.info.assert_called_once()
+        entry = json.loads(mock_audit.info.call_args[0][0])
         assert entry["agent_name"] == "test_agent"
         assert entry["user_id"] == "user-001"
         assert entry["script_length"] == len("x = 42")

tests/tools/safety/test_integration.py

@@ -16,7 +16,6 @@
 
 from __future__ import annotations
 
-import logging
 from pathlib import Path
 from unittest.mock import AsyncMock, MagicMock, patch
 
@@ -382,26 +381,24 @@ def test_hardcoded_secret_bash(self, default_guard: ScriptSafetyGuard):
 class TestAuditAndTelemetry:
     """Verify audit logging and OTel recording fires correctly."""
 
-    def test_audit_log_emitted(self, default_guard: ScriptSafetyGuard, caplog):
+    def test_audit_log_emitted(self, default_guard: ScriptSafetyGuard):
         """Guard should emit structured audit log on every check."""
         code = 'print("hello")'
-        with caplog.at_level(logging.INFO, logger="trpc_agent_sdk.tools.safety.audit"):
+        with patch("trpc_agent_sdk.tools.safety.guard._audit_logger") as mock_audit:
             default_guard.check(_make_input(code))
 
-        # Verify audit log was emitted
-        audit_records = [r for r in caplog.records if r.name == "trpc_agent_sdk.tools.safety.audit"]
-        assert len(audit_records) == 1
-        assert "safety_check" in audit_records[0].getMessage()
+        mock_audit.info.assert_called_once()
+        msg = mock_audit.info.call_args[0][0]
+        assert "safety_check" in msg
 
-    def test_audit_log_contains_decision(self, default_guard: ScriptSafetyGuard, caplog):
+    def test_audit_log_contains_decision(self, default_guard: ScriptSafetyGuard):
         """Audit log should contain decision and findings info."""
         code = 'result = eval("1+1")'
-        with caplog.at_level(logging.INFO, logger="trpc_agent_sdk.tools.safety.audit"):
+        with patch("trpc_agent_sdk.tools.safety.guard._audit_logger") as mock_audit:
             default_guard.check(_make_input(code))
 
-        audit_records = [r for r in caplog.records if r.name == "trpc_agent_sdk.tools.safety.audit"]
-        assert len(audit_records) == 1
-        msg = audit_records[0].getMessage()
+        mock_audit.info.assert_called_once()
+        msg = mock_audit.info.call_args[0][0]
         assert '"decision": "deny"' in msg
         assert '"PROC-002"' in msg
 

trpc_agent_sdk/tools/safety/adapters/filter_adapter.py

@@ -18,7 +18,7 @@
 import logging
 from typing import Any, Optional
 
-from trpc_agent_sdk.abc import FilterResult, FilterType
+from trpc_agent_sdk.abc import FilterResult
 from trpc_agent_sdk.context import AgentContext
 from trpc_agent_sdk.filter import BaseFilter, register_tool_filter
 from trpc_agent_sdk.tools.safety.guard import ScriptSafetyGuard

trpc_agent_sdk/tools/safety/rules/file_ops.py

@@ -7,7 +7,6 @@
 
 from __future__ import annotations
 
-import ast
 import fnmatch
 import os
 from typing import TYPE_CHECKING

trpc_agent_sdk/tools/safety/scanner/bash_scanner.py

@@ -7,7 +7,7 @@
 from __future__ import annotations
 
 import re
-from dataclasses import dataclass, field
+from dataclasses import dataclass
 from typing import Optional