feat(tools): add safety audit logger with JSONL output

Author: DNKYr

tests/tools/safety/test_audit.py

@@ -0,0 +1,105 @@
+# Tencent is pleased to support the open source community by making tRPC-Agent-Python available.
+#
+# Copyright (C) 2026 Tencent. All rights reserved.
+#
+# tRPC-Agent-Python is licensed under Apache-2.0.
+"""Tests for audit logging."""
+
+import json
+import tempfile
+from pathlib import Path
+
+from trpc_agent_sdk.tools.safety._audit import SafetyAuditLogger
+from trpc_agent_sdk.tools.safety._types import AuditEvent
+from trpc_agent_sdk.tools.safety._types import Decision
+from trpc_agent_sdk.tools.safety._types import ScanReport
+
+
+class TestSafetyAuditLogger:
+    def test_log_writes_valid_jsonl(self):
+        with tempfile.NamedTemporaryFile(mode="w", suffix=".jsonl", delete=False) as f:
+            log_path = f.name
+        try:
+            logger = SafetyAuditLogger(output_path=log_path)
+            event = AuditEvent(
+                timestamp="2026-07-10T12:00:00Z",
+                tool_name="bash_tool",
+                decision="deny",
+                risk_level="critical",
+                rule_ids=["DANGEROUS_DELETE_001"],
+                scan_duration_ms=12.5,
+                sanitized=False,
+                intercepted=True,
+                script_hash="abc123",
+            )
+            logger.log_event(event)
+
+            with open(log_path) as f:
+                lines = f.readlines()
+            assert len(lines) == 1
+            parsed = json.loads(lines[0])
+            assert parsed["tool_name"] == "bash_tool"
+            assert parsed["decision"] == "deny"
+        finally:
+            Path(log_path).unlink()
+
+    def test_log_from_scan_report(self):
+        with tempfile.NamedTemporaryFile(mode="w", suffix=".jsonl", delete=False) as f:
+            log_path = f.name
+        try:
+            logger = SafetyAuditLogger(output_path=log_path)
+            report = ScanReport(
+                decision=Decision.ALLOW,
+                findings=[],
+                scan_duration_ms=3.1,
+            )
+            logger.log(report, tool_name="python_tool", script_hash="def456")
+
+            with open(log_path) as f:
+                lines = f.readlines()
+            assert len(lines) == 1
+            parsed = json.loads(lines[0])
+            assert parsed["decision"] == "allow"
+            assert parsed["intercepted"] is False
+            assert parsed["risk_level"] is None
+            assert parsed["tool_name"] == "python_tool"
+        finally:
+            Path(log_path).unlink()
+
+    def test_multiple_events_written(self):
+        with tempfile.NamedTemporaryFile(mode="w", suffix=".jsonl", delete=False) as f:
+            log_path = f.name
+        try:
+            logger = SafetyAuditLogger(output_path=log_path)
+            logger.log_event(AuditEvent(
+                timestamp="2026-07-10T12:00:00Z",
+                tool_name="tool_a",
+                decision="allow",
+                risk_level=None,
+                rule_ids=[],
+                scan_duration_ms=1.0,
+                sanitized=False,
+                intercepted=False,
+                script_hash="aaa",
+            ))
+            logger.log_event(AuditEvent(
+                timestamp="2026-07-10T12:01:00Z",
+                tool_name="tool_b",
+                decision="deny",
+                risk_level="high",
+                rule_ids=["NETWORK_PYTHON_004"],
+                scan_duration_ms=2.0,
+                sanitized=False,
+                intercepted=True,
+                script_hash="bbb",
+            ))
+
+            with open(log_path) as f:
+                lines = f.readlines()
+            assert len(lines) == 2
+            parsed_a = json.loads(lines[0])
+            parsed_b = json.loads(lines[1])
+            assert parsed_a["tool_name"] == "tool_a"
+            assert parsed_b["tool_name"] == "tool_b"
+        finally:
+            Path(log_path).unlink()

trpc_agent_sdk/tools/safety/_audit.py

@@ -0,0 +1,53 @@
+# Tencent is pleased to support the open source community by making tRPC-Agent-Python available.
+#
+# Copyright (C) 2026 Tencent. All rights reserved.
+#
+# tRPC-Agent-Python is licensed under Apache-2.0.
+"""Audit logging for the Tool Script Safety Guard.
+
+Writes structured audit events as JSONL (one JSON object per line).
+"""
+
+from __future__ import annotations
+
+import json
+from dataclasses import asdict
+from datetime import datetime
+from datetime import timezone
+from pathlib import Path
+from typing import Optional
+
+from ._types import AuditEvent
+from ._types import ScanReport
+
+
+class SafetyAuditLogger:
+    """Logs safety scan events to a JSONL file."""
+
+    def __init__(self, output_path: str = "tool_safety_audit.jsonl"):
+        self.output_path = Path(output_path)
+        self.output_path.parent.mkdir(parents=True, exist_ok=True)
+
+    def log_event(self, event: AuditEvent):
+        with open(self.output_path, "a", encoding="utf-8") as f:
+            f.write(json.dumps(asdict(event), ensure_ascii=False) + "\n")
+
+    def log(
+        self,
+        report: ScanReport,
+        tool_name: str,
+        script_hash: str,
+        sanitized: bool = False,
+    ):
+        event = AuditEvent(
+            timestamp=datetime.now(timezone.utc).isoformat(),
+            tool_name=tool_name,
+            decision=report.decision.value,
+            risk_level=report.risk_level.value if report.risk_level else None,
+            rule_ids=[f.rule_id for f in report.findings],
+            scan_duration_ms=report.scan_duration_ms,
+            sanitized=sanitized,
+            intercepted=(report.decision.value == "deny"),
+            script_hash=script_hash,
+        )
+        self.log_event(event)