Coverage for src / lexigram / contracts / ai / guards.py: 0%

1"""AI guard chain protocols for input/output policy enforcement.""" 

2 

3from __future__ import annotations 

4 

5from typing import TYPE_CHECKING, Any, Protocol, runtime_checkable 

6 

7from lexigram.contracts.ai.exceptions import GuardError 

8 

9if TYPE_CHECKING: 

10 from lexigram.contracts.core.result import Result 

11 

12 

13# Guard Errors 

14class InputGuardError(GuardError): 

15 """Error raised during input guard validation.""" 

16 

17 _code = "LEX_ERR_GUARD_002" 

18 

19 

20class OutputGuardError(GuardError): 

21 """Error raised during output guard validation.""" 

22 

23 _code = "LEX_ERR_GUARD_003" 

24 

25 

26@runtime_checkable 

27class GuardResultProtocol(Protocol): 

28 """Protocol for guard evaluation results. 

29 

30 Every guard returns a result indicating whether the content 

31 passed, was blocked, or triggered a warning. 

32 """ 

33 

34 @property 

35 def passed(self) -> bool: 

36 """Whether the guard check passed.""" 

37 ... 

38 

39 @property 

40 def action(self) -> str: 

41 """Action taken: 'pass', 'block', 'warn', or 'redact'.""" 

42 ... 

43 

44 @property 

45 def guard_name(self) -> str: 

46 """Name of the guard that produced this result.""" 

47 ... 

48 

49 @property 

50 def details(self) -> dict[str, Any]: 

51 """Additional details about the guard evaluation.""" 

52 ... 

53 

54 @property 

55 def redacted_content(self) -> str | None: 

56 """Redacted content, if action was 'redact'.""" 

57 ... 

58 

59 

60@runtime_checkable 

61class InputGuardProtocol(Protocol): 

62 """Protocol for input content guards. 

63 

64 Input guards inspect user prompts and messages before they are 

65 sent to an LLM. They can block, warn, or redact content. 

66 """ 

67 

68 @property 

69 def name(self) -> str: 

70 """GuardProtocol identifier.""" 

71 ... 

72 

73 async def check( 

74 self, 

75 content: str, 

76 *, 

77 messages: list[Any] | None = None, 

78 metadata: dict[str, Any] | None = None, 

79 ) -> Result[GuardResultProtocol, GuardError]: 

80 """Evaluate input content against this guard's rules. 

81 

82 Args: 

83 content: The raw text content to check. 

84 messages: Optional structured message list for context. 

85 metadata: Optional metadata (user_id, model, etc.). 

86 

87 Returns: 

88 A GuardCheckResult indicating pass/block/warn/redact. 

89 """ 

90 ... 

91 

92 

93@runtime_checkable 

94class OutputGuardProtocol(Protocol): 

95 """Protocol for output content guards. 

96 

97 Output guards inspect LLM responses before they are returned 

98 to the caller. They can block, warn, or redact content. 

99 """ 

100 

101 @property 

102 def name(self) -> str: 

103 """GuardProtocol identifier.""" 

104 ... 

105 

106 async def check( 

107 self, 

108 content: str, 

109 *, 

110 original_input: str | None = None, 

111 metadata: dict[str, Any] | None = None, 

112 ) -> Result[GuardResultProtocol, GuardError]: 

113 """Evaluate output content against this guard's rules. 

114 

115 Args: 

116 content: The LLM response text to check. 

117 original_input: The original user input for context. 

118 metadata: Optional metadata (model, provider, etc.). 

119 

120 Returns: 

121 A GuardCheckResult indicating pass/block/warn/redact. 

122 """ 

123 ... 

124 

125 

126@runtime_checkable 

127class GuardPipelineProtocol(Protocol): 

128 """Protocol for guard pipeline execution. 

129 

130 Orchestrates a chain of input and/or output guards, collecting 

131 results and determining the aggregate action. 

132 """ 

133 

134 async def check_input( 

135 self, 

136 content: str, 

137 *, 

138 messages: list[Any] | None = None, 

139 metadata: dict[str, Any] | None = None, 

140 ) -> Result[GuardResultProtocol, GuardError]: 

141 """Run all input guards against the content. 

142 

143 Args: 

144 content: Input text to guard. 

145 messages: Optional structured messages. 

146 metadata: Optional request metadata. 

147 

148 Returns: 

149 Aggregate guard result. 

150 """ 

151 ... 

152 

153 async def check_output( 

154 self, 

155 content: str, 

156 *, 

157 original_input: str | None = None, 

158 metadata: dict[str, Any] | None = None, 

159 ) -> Result[GuardResultProtocol, GuardError]: 

160 """Run all output guards against the content. 

161 

162 Args: 

163 content: Output text to guard. 

164 original_input: The original user input. 

165 metadata: Optional request metadata. 

166 

167 Returns: 

168 Aggregate guard result. 

169 """ 

170 ... 

171 

172 

173__all__ = [ 

174 "GuardError", 

175 "GuardPipelineProtocol", 

176 "GuardResultProtocol", 

177 "InputGuardError", 

178 "InputGuardProtocol", 

179 "OutputGuardError", 

180 "OutputGuardProtocol", 

181]