test(safety): add types unit tests for tool safety scanner

16 tests covering Decision, RiskLevel, Finding, Report, AuditEvent,
Policy, Request, and CodeBlock types. Enums with rank ordering.

Mirrors trpc-agent-go/tool/safety/safety.go types.

Signed-off-by: coder-mtj <coder-mtj@users.noreply.github.com>

Author: coder-mtj

tests/tools/safety/__init__.py

@@ -0,0 +1 @@
+# Tool safety guard tests.

tests/tools/safety/test_types.py

@@ -0,0 +1,221 @@
+# 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.
+
+"""Unit tests for safety scanner type definitions.
+
+Mirrors trpc-agent-go/tool/safety/safety.go types.
+"""
+
+from __future__ import annotations
+
+import dataclasses
+import json
+import pytest
+
+
+class TestDecision:
+    def test_decision_constants(self):
+        from trpc_agent_sdk.tools.safety._types import (
+            Decision, DECISION_ALLOW, DECISION_DENY,
+            DECISION_ASK, DECISION_NEEDS_HUMAN_REVIEW,
+        )
+        assert Decision(DECISION_ALLOW) == Decision("allow")
+        assert Decision(DECISION_DENY) == Decision("deny")
+        assert Decision(DECISION_ASK) == Decision("ask")
+        assert Decision(DECISION_NEEDS_HUMAN_REVIEW) == Decision("needs_human_review")
+
+    def test_decision_rank_order(self):
+        from trpc_agent_sdk.tools.safety._types import (
+            DECISION_ALLOW, DECISION_ASK, DECISION_NEEDS_HUMAN_REVIEW,
+            DECISION_DENY, decision_rank,
+        )
+        assert decision_rank(DECISION_DENY) > decision_rank(DECISION_NEEDS_HUMAN_REVIEW)
+        assert decision_rank(DECISION_NEEDS_HUMAN_REVIEW) > decision_rank(DECISION_ASK)
+        assert decision_rank(DECISION_ASK) > decision_rank(DECISION_ALLOW)
+
+    def test_decision_str_value(self):
+        from trpc_agent_sdk.tools.safety._types import DECISION_ALLOW
+        assert DECISION_ALLOW.value == "allow"
+
+
+class TestRiskLevel:
+    def test_risk_level_constants(self):
+        from trpc_agent_sdk.tools.safety._types import (
+            RISK_LOW, RISK_MEDIUM, RISK_HIGH, RISK_CRITICAL,
+        )
+        assert RISK_LOW.value == "low"
+        assert RISK_MEDIUM.value == "medium"
+        assert RISK_HIGH.value == "high"
+        assert RISK_CRITICAL.value == "critical"
+
+    def test_risk_rank_order(self):
+        from trpc_agent_sdk.tools.safety._types import (
+            RISK_LOW, RISK_MEDIUM, RISK_HIGH, RISK_CRITICAL, risk_rank,
+        )
+        assert risk_rank(RISK_CRITICAL) > risk_rank(RISK_HIGH)
+        assert risk_rank(RISK_HIGH) > risk_rank(RISK_MEDIUM)
+        assert risk_rank(RISK_MEDIUM) > risk_rank(RISK_LOW)
+
+
+class TestFinding:
+    def test_finding_fields(self):
+        from trpc_agent_sdk.tools.safety._types import Finding, DECISION_DENY, RISK_CRITICAL
+        f = Finding(
+            decision=DECISION_DENY,
+            risk_level=RISK_CRITICAL,
+            rule_id="test.rule",
+            evidence=["/etc/passwd"],
+            recommendation="Do not access system files.",
+        )
+        assert f.decision == DECISION_DENY
+        assert f.rule_id == "test.rule"
+        assert f.evidence == ["/etc/passwd"]
+
+    def test_finding_beats_by_decision(self):
+        from trpc_agent_sdk.tools.safety._types import (
+            Finding, finding_beats, DECISION_DENY, DECISION_ALLOW,
+            RISK_LOW, RISK_CRITICAL,
+        )
+        bad = Finding(DECISION_DENY, RISK_LOW, "r1", ["ev"], "rec")
+        good = Finding(DECISION_ALLOW, RISK_CRITICAL, "r2", ["ev"], "rec")
+        assert finding_beats(bad, good), "deny should beat allow regardless of risk"
+
+    def test_finding_beats_by_risk_when_same_decision(self):
+        from trpc_agent_sdk.tools.safety._types import (
+            Finding, finding_beats, DECISION_DENY,
+            RISK_LOW, RISK_CRITICAL,
+        )
+        critical = Finding(DECISION_DENY, RISK_CRITICAL, "r1", ["ev"], "rec")
+        low = Finding(DECISION_DENY, RISK_LOW, "r2", ["ev"], "rec")
+        assert finding_beats(critical, low), "critical should beat low for same decision"
+
+
+class TestReport:
+    def test_report_serialization(self):
+        from trpc_agent_sdk.tools.safety._types import (
+            Report, Finding, DECISION_ALLOW, RISK_LOW,
+        )
+        r = Report(
+            decision=DECISION_ALLOW,
+            risk_level=RISK_LOW,
+            recommendation="Safe.",
+            tool_name="test_tool",
+            command="echo hi",
+            backend="workspaceexec",
+            blocked=False,
+            duration_ms=5,
+            redacted=False,
+            findings=[],
+        )
+        data = dataclasses.asdict(r)
+        assert data["decision"] == "allow"
+        assert data["blocked"] == False
+
+    def test_report_blocked_if_deny(self):
+        from trpc_agent_sdk.tools.safety._types import (
+            Report, DECISION_DENY, RISK_HIGH,
+        )
+        r = Report(
+            decision=DECISION_DENY,
+            risk_level=RISK_HIGH,
+            recommendation="Blocked.",
+            tool_name="test",
+            command="rm -rf /",
+            backend="workspaceexec",
+            blocked=True,
+            duration_ms=1,
+            redacted=False,
+        )
+        assert r.blocked is True
+
+    def test_span_attributes(self):
+        from trpc_agent_sdk.tools.safety._types import (
+            Report, DECISION_DENY, RISK_CRITICAL,
+        )
+        r = Report(
+            decision=DECISION_DENY,
+            risk_level=RISK_CRITICAL,
+            rule_id="dangerous.rm_rf",
+            recommendation="Denied.",
+            tool_name="ws",
+            command="rm -rf /",
+            backend="workspaceexec",
+            blocked=True,
+            duration_ms=2,
+            redacted=False,
+        )
+        attrs = r.span_attributes()
+        assert attrs["tool.safety.decision"] == "deny"
+        assert attrs["tool.safety.risk_level"] == "critical"
+        assert attrs["tool.safety.rule_id"] == "dangerous.rm_rf"
+
+
+class TestAuditEvent:
+    def test_audit_event_jsonl(self):
+        from trpc_agent_sdk.tools.safety._types import (
+            AuditEvent, DECISION_DENY, RISK_HIGH,
+        )
+        import time
+        evt = AuditEvent(
+            timestamp=time.time(),
+            tool_name="test",
+            decision=DECISION_DENY,
+            risk_level=RISK_HIGH,
+            rule_id="r1",
+            duration_ms=5,
+            redacted=False,
+            blocked=True,
+            backend="workspaceexec",
+        )
+        line = json.dumps(dataclasses.asdict(evt), default=str)
+        data = json.loads(line)
+        assert data["decision"] == "deny"
+        assert data["blocked"] == True
+
+
+class TestPolicy:
+    def test_policy_defaults(self):
+        from trpc_agent_sdk.tools.safety._types import Policy
+        p = Policy()
+        assert p.max_timeout_seconds == 0  # default before DefaultPolicy()
+        assert p.max_output_bytes == 0
+
+    def test_policy_fields(self):
+        from trpc_agent_sdk.tools.safety._types import Policy
+        p = Policy(
+            denied_commands=["rm", "sudo"],
+            denied_paths=["/etc", ".ssh"],
+            network_allowlist=["api.github.com"],
+            max_timeout_seconds=300,
+        )
+        assert "rm" in p.denied_commands
+        assert "/etc" in p.denied_paths
+        assert "api.github.com" in p.network_allowlist
+
+
+class TestRequest:
+    def test_request_fields(self):
+        from trpc_agent_sdk.tools.safety._types import Request
+        r = Request(
+            tool_name="workspace_exec",
+            command="go test ./...",
+            cwd=".",
+            backend="workspaceexec",
+            timeout_seconds=30,
+        )
+        assert r.tool_name == "workspace_exec"
+        assert r.command == "go test ./..."
+
+    def test_request_with_code_blocks(self):
+        from trpc_agent_sdk.tools.safety._types import Request, CodeBlock
+        r = Request(
+            tool_name="execute_code",
+            backend="codeexec",
+            code_blocks=[CodeBlock(language="python", code="print(1)")],
+        )
+        assert len(r.code_blocks) == 1
+        assert r.code_blocks[0].language == "python"

trpc_agent_sdk/tools/safety/__init__.py

@@ -0,0 +1,4 @@
+# Package safety provides a pre-execution safety guard for command and
+# code-execution tools.
+#
+# Mirrors trpc-agent-go/tool/safety/.

trpc_agent_sdk/tools/safety/_types.py

@@ -0,0 +1,168 @@
+# 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.
+
+"""Safety scanner data types. Mirrors trpc-agent-go/tool/safety/safety.go."""
+
+from __future__ import annotations
+
+import time
+from dataclasses import dataclass, field
+from enum import Enum
+
+
+# ---------------------------------------------------------------------------
+# Decision & RiskLevel
+# ---------------------------------------------------------------------------
+
+class Decision(str, Enum):
+    ALLOW = "allow"
+    DENY = "deny"
+    ASK = "ask"
+    NEEDS_HUMAN_REVIEW = "needs_human_review"
+
+
+DECISION_ALLOW = Decision.ALLOW
+DECISION_DENY = Decision.DENY
+DECISION_ASK = Decision.ASK
+DECISION_NEEDS_HUMAN_REVIEW = Decision.NEEDS_HUMAN_REVIEW
+
+
+class RiskLevel(str, Enum):
+    LOW = "low"
+    MEDIUM = "medium"
+    HIGH = "high"
+    CRITICAL = "critical"
+
+
+RISK_LOW = RiskLevel.LOW
+RISK_MEDIUM = RiskLevel.MEDIUM
+RISK_HIGH = RiskLevel.HIGH
+RISK_CRITICAL = RiskLevel.CRITICAL
+
+
+def decision_rank(d: Decision) -> int:
+    _map = {
+        Decision.ALLOW: 1,
+        Decision.ASK: 2,
+        Decision.NEEDS_HUMAN_REVIEW: 3,
+        Decision.DENY: 4,
+    }
+    return _map.get(d, 0)
+
+
+def risk_rank(level: RiskLevel) -> int:
+    _map = {
+        RiskLevel.LOW: 1,
+        RiskLevel.MEDIUM: 2,
+        RiskLevel.HIGH: 3,
+        RiskLevel.CRITICAL: 4,
+    }
+    return _map.get(level, 0)
+
+
+# ---------------------------------------------------------------------------
+# Policy
+# ---------------------------------------------------------------------------
+
+@dataclass
+class Policy:
+    denied_commands: list[str] = field(default_factory=list)
+    allowed_commands: list[str] = field(default_factory=list)
+    denied_paths: list[str] = field(default_factory=list)
+    network_allowlist: list[str] = field(default_factory=list)
+    env_allowlist: list[str] = field(default_factory=list)
+    review_commands: list[str] = field(default_factory=list)
+    max_timeout_seconds: int = 0
+    max_output_bytes: int = 0
+    review_shell_pipelines: bool = True
+    deny_on_parse_error: bool = True
+
+
+# ---------------------------------------------------------------------------
+# Request
+# ---------------------------------------------------------------------------
+
+@dataclass
+class CodeBlock:
+    language: str = ""
+    code: str = ""
+
+
+@dataclass
+class Request:
+    tool_name: str = ""
+    command: str = ""
+    args: list[str] = field(default_factory=list)
+    cwd: str = ""
+    env: dict[str, str] = field(default_factory=dict)
+    backend: str = ""
+    timeout_seconds: int = 0
+    max_output_bytes: int = 0
+    background: bool = False
+    tty: bool = False
+    code_blocks: list[CodeBlock] = field(default_factory=list)
+
+
+# ---------------------------------------------------------------------------
+# Finding & Report
+# ---------------------------------------------------------------------------
+
+@dataclass
+class Finding:
+    decision: Decision = Decision.ALLOW
+    risk_level: RiskLevel = RiskLevel.LOW
+    rule_id: str = ""
+    evidence: list[str] = field(default_factory=list)
+    recommendation: str = ""
+
+
+def finding_beats(a: Finding, b: Finding) -> bool:
+    if decision_rank(a.decision) != decision_rank(b.decision):
+        return decision_rank(a.decision) > decision_rank(b.decision)
+    return risk_rank(a.risk_level) > risk_rank(b.risk_level)
+
+
+@dataclass
+class Report:
+    decision: Decision = Decision.ALLOW
+    risk_level: RiskLevel = RiskLevel.LOW
+    rule_id: str = ""
+    evidence: list[str] = field(default_factory=list)
+    recommendation: str = ""
+    tool_name: str = ""
+    command: str = ""
+    backend: str = ""
+    blocked: bool = False
+    redacted: bool = False
+    duration_ms: int = 0
+    safe_summary: str = ""
+    findings: list[Finding] = field(default_factory=list)
+
+    def span_attributes(self) -> dict[str, str]:
+        return {
+            "tool.safety.decision": self.decision.value,
+            "tool.safety.risk_level": self.risk_level.value,
+            "tool.safety.rule_id": self.rule_id,
+            "tool.safety.backend": self.backend,
+        }
+
+
+# ---------------------------------------------------------------------------
+# AuditEvent
+# ---------------------------------------------------------------------------
+
+@dataclass
+class AuditEvent:
+    timestamp: float = field(default_factory=time.time)
+    tool_name: str = ""
+    decision: Decision = Decision.ALLOW
+    risk_level: RiskLevel = RiskLevel.LOW
+    rule_id: str = ""
+    duration_ms: int = 0
+    redacted: bool = False
+    blocked: bool = False
+    backend: str = ""