Coverage for /home/admin/Documents/AI/applications/lexigram-dev/lexigram/experimental/ai/lexigram-ai-guard/src/lexigram/ai/guard/pipeline/result.py: 75%
72 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-25 07:19 +0800
1"""GuardProtocol result types.
3Immutable value objects representing the outcome of a single guard
4or an aggregate pipeline evaluation.
5"""
7from __future__ import annotations
9from dataclasses import dataclass, field
10from enum import StrEnum
11from typing import TYPE_CHECKING, Any
13if TYPE_CHECKING:
14 from lexigram.contracts.ai.guards import GuardResultProtocol
17class GuardAction(StrEnum):
18 """Action to take on a guard result."""
20 PASS = "pass" # noqa: S105 # guard verdict enum, not a credential
21 """Content is safe — allow through."""
23 BLOCK = "block"
24 """Content is unsafe — reject the request."""
26 WARN = "warn"
27 """Content is borderline — allow but emit a warning."""
29 REDACT = "redact"
30 """Content contains sensitive information — redact before allowing."""
33@dataclass(frozen=True)
34class GuardCheckResult:
35 """Result of a single guard evaluation.
37 Immutable result produced by one :class:`InputGuardProtocol` or
38 :class:`OutputGuardProtocol` check.
39 """
41 guard_name: str
42 """Identifier of the guard that produced this result."""
44 passed: bool
45 """Whether the guard check passed (action is PASS or WARN)."""
47 action: str
48 """Action to take: 'pass', 'block', 'warn', or 'redact'."""
50 details: dict[str, Any] = field(default_factory=dict)
51 """Additional diagnostic details from the guard evaluation."""
53 redacted_content: str | None = None
54 """Redacted version of the content, if action is 'redact'."""
56 @classmethod
57 def allow(cls, guard_name: str, **details: Any) -> GuardCheckResult:
58 """Create a passing guard result.
60 Args:
61 guard_name: GuardProtocol identifier.
62 **details: Optional extra diagnostics.
64 Returns:
65 GuardCheckResult with action=PASS.
66 """
67 return cls(
68 guard_name=guard_name,
69 passed=True,
70 action=GuardAction.PASS,
71 details=dict(details),
72 )
74 @classmethod
75 def block(cls, guard_name: str, reason: str, **details: Any) -> GuardCheckResult:
76 """Create a blocking guard result.
78 Args:
79 guard_name: GuardProtocol identifier.
80 reason: Human-readable explanation of why the content was blocked.
81 **details: Optional extra diagnostics.
83 Returns:
84 GuardCheckResult with action=BLOCK.
85 """
86 return cls(
87 guard_name=guard_name,
88 passed=False,
89 action=GuardAction.BLOCK,
90 details={"reason": reason, **details},
91 )
93 @classmethod
94 def warn(cls, guard_name: str, reason: str, **details: Any) -> GuardCheckResult:
95 """Create a warning guard result.
97 Args:
98 guard_name: GuardProtocol identifier.
99 reason: Human-readable explanation of why a warning was emitted.
100 **details: Optional extra diagnostics.
102 Returns:
103 GuardCheckResult with action=WARN (passed=True).
104 """
105 return cls(
106 guard_name=guard_name,
107 passed=True,
108 action=GuardAction.WARN,
109 details={"reason": reason, **details},
110 )
112 @classmethod
113 def redact(
114 cls,
115 guard_name: str,
116 redacted_content: str,
117 reason: str,
118 **details: Any,
119 ) -> GuardCheckResult:
120 """Create a redacting guard result.
122 Args:
123 guard_name: GuardProtocol identifier.
124 redacted_content: The sanitized version of the original content.
125 reason: Human-readable explanation of what was redacted.
126 **details: Optional extra diagnostics.
128 Returns:
129 GuardCheckResult with action=REDACT (passed=True).
130 """
131 return cls(
132 guard_name=guard_name,
133 passed=True,
134 action=GuardAction.REDACT,
135 details={"reason": reason, **details},
136 redacted_content=redacted_content,
137 )
140@dataclass(frozen=True)
141class AggregateGuardResult:
142 """Aggregate result from running multiple guards in a pipeline.
144 Combines results from all guards into a single verdict.
145 The aggregate action is the most severe action from any guard
146 (BLOCK > REDACT > WARN > PASS).
147 """
149 passed: bool
150 """Whether all guards passed (no BLOCK action)."""
152 action: str
153 """Most severe action from any guard (BLOCK > REDACT > WARN > PASS)."""
155 results: list[GuardResultProtocol] = field(default_factory=list)
156 """Individual results from each guard, in evaluation order."""
158 final_content: str | None = None
159 """Content after redaction (may differ from input if any guard redacted)."""
161 @property
162 def blocked(self) -> bool:
163 """Whether any guard triggered a block action."""
164 return self.action == GuardAction.BLOCK
166 @property
167 def redacted(self) -> bool:
168 """Whether any guard redacted content."""
169 return self.action == GuardAction.REDACT
171 @property
172 def warned(self) -> bool:
173 """Whether any guard emitted a warning."""
174 return self.action == GuardAction.WARN
176 @property
177 def blocking_result(self) -> GuardResultProtocol | None:
178 """Return the first blocking result, or None if none blocked."""
179 return next((r for r in self.results if r.action == GuardAction.BLOCK), None)
181 @classmethod
182 def from_results(
183 cls,
184 results: list[GuardResultProtocol],
185 original_content: str,
186 ) -> AggregateGuardResult:
187 """Build an aggregate result from a list of individual results.
189 The aggregate action uses the most severe outcome. If multiple
190 guards redacted content, only the last redaction is applied (guards
191 should be ordered most-restrictive-first in the pipeline).
193 Args:
194 results: Individual guard check results.
195 original_content: Original input content before any redaction.
197 Returns:
198 Aggregated result.
199 """
200 _severity: dict[str, int] = {
201 GuardAction.BLOCK: 3,
202 GuardAction.REDACT: 2,
203 GuardAction.WARN: 1,
204 GuardAction.PASS: 0,
205 }
207 final_action = GuardAction.PASS
208 final_content = original_content
210 for result in results:
211 if _severity.get(result.action, 0) > _severity.get(final_action, 0):
212 final_action = GuardAction(result.action)
213 if (
214 result.action == GuardAction.REDACT
215 and result.redacted_content is not None
216 ):
217 final_content = result.redacted_content
219 passed = final_action != GuardAction.BLOCK
221 return cls(
222 passed=passed,
223 action=final_action,
224 results=results,
225 final_content=final_content if final_action != GuardAction.BLOCK else None,
226 )
229__all__ = [
230 "AggregateGuardResult",
231 "GuardAction",
232 "GuardCheckResult",
233]