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

1"""AI governance contracts.""" 

2 

3from __future__ import annotations 

4 

5from dataclasses import dataclass, field 

6from datetime import UTC, datetime 

7from enum import StrEnum 

8from typing import Any, Protocol, runtime_checkable 

9from uuid import uuid4 

10 

11from lexigram.contracts.ai.governance.errors import ( 

12 BudgetExceededError, 

13 GovernanceError, 

14 PolicyViolationError, 

15) 

16from lexigram.contracts.ai.governance.relay_billing import ( 

17 RelayBillingError, 

18 RelayBillingProtocol, 

19 RelayChargeBreakdown, 

20 RelayPriceEstimatorProtocol, 

21 RelayUsageRecord, 

22 RelayUsageReservation, 

23 RelayUsageScope, 

24 RelayUsageStoreProtocol, 

25 billing_store_unavailable, 

26 charge_overflow, 

27 duplicate_settlement, 

28 invalid_usage, 

29 quota_exhausted, 

30 reservation_expired, 

31 unknown_price, 

32) 

33from lexigram.contracts.ai.governance.resource_unit import ( 

34 ResourceExhaustedError, 

35 ResourceQuota, 

36 ResourceUnit, 

37 ResourceUsageResult, 

38 ResourceUsageSnapshot, 

39 ResourceWindowKind, 

40) 

41 

42 

43@dataclass(frozen=True) 

44class GovernanceDecision: 

45 """Decision from the governance layer.""" 

46 

47 allowed: bool 

48 reason: str | None = None 

49 remaining_budget: float | None = None 

50 

51 

52@runtime_checkable 

53class CostTrackingProtocol(Protocol): 

54 """Protocol for tracking costs.""" 

55 

56 async def track_cost( 

57 self, 

58 cost: float, 

59 model: str, 

60 user_id: str | None = None, 

61 ) -> None: 

62 """Track cost for an operation.""" 

63 ... 

64 

65 async def get_budget(self, user_id: str | None = None) -> float: 

66 """Get remaining budget.""" 

67 ... 

68 

69 

70@runtime_checkable 

71class AIGovernanceProtocol(Protocol): 

72 """Protocol for AI Governance and cost tracking.""" 

73 

74 async def check_request( 

75 self, 

76 model: str, 

77 provider: str, 

78 user_id: str | None = None, 

79 ) -> GovernanceDecision: 

80 """Check if request is allowed by governance policies.""" 

81 ... 

82 

83 async def check_budget(self, cost: float, user_id: str | None = None) -> bool: 

84 """Check if operation fits within budget.""" 

85 ... 

86 

87 async def track_cost( 

88 self, 

89 cost: float, 

90 model: str, 

91 user_id: str | None = None, 

92 ) -> None: 

93 """Track cost for an operation.""" 

94 ... 

95 

96 

97class AuditEventType(StrEnum): 

98 """Categories of auditable AI operations. 

99 

100 Shared across the AI governance and observability layers 

101 via ``lexigram-contracts``. 

102 """ 

103 

104 CREATED = "created" 

105 LLM_CALL = "llm_call" 

106 MODEL_DENIED = "model_denied" 

107 BUDGET_EXCEEDED = "budget_exceeded" 

108 RATE_LIMITED = "rate_limited" 

109 TOOL_CALL = "tool_call" 

110 AGENT_EXECUTION = "agent_execution" 

111 SOFT_LIMIT_REACHED = "soft_limit_reached" 

112 CONFIG_RELOADED = "config_reloaded" 

113 

114 

115@dataclass(frozen=True) 

116class AIAuditEvent: 

117 """Structured representation of an auditable AI operation. 

118 

119 Concrete data class shared across AI packages via ``lexigram-contracts``. 

120 

121 Attributes: 

122 event_type: Category of the operation. 

123 model: Model identifier (if applicable). 

124 provider: Provider name (if applicable). 

125 user_id: User who triggered the operation. 

126 status: Outcome — ``"allowed"``, ``"denied"``, ``"success"``, ``"error"``. 

127 tokens: Token count consumed (if applicable). 

128 cost: Dollar cost incurred (if applicable). 

129 latency_ms: Request latency in milliseconds (if applicable). 

130 metadata: Free-form key/value bag for additional context. 

131 event_id: Unique UUID for the event (auto-generated). 

132 """ 

133 

134 event_type: AuditEventType 

135 model: str | None = None 

136 provider: str | None = None 

137 user_id: str | None = None 

138 status: str = "success" 

139 tokens: int | None = None 

140 cost: float | None = None 

141 latency_ms: float | None = None 

142 metadata: dict[str, Any] = field(default_factory=dict) 

143 event_id: str = field(default_factory=lambda: str(uuid4())) 

144 timestamp: datetime = field(default_factory=lambda: datetime.now(UTC)) 

145 

146 

147@runtime_checkable 

148class AIAuditStoreProtocol(Protocol): 

149 """Protocol for AI audit event persistence backends. 

150 

151 Implementations must be async and treat ``record()`` as 

152 fire-and-forget safe — hot-paths must never block on audit persistence. 

153 """ 

154 

155 async def record(self, event: AIAuditEvent) -> None: 

156 """Persist a single audit event. 

157 

158 Args: 

159 event: The audit event to store. 

160 """ 

161 ... 

162 

163 

164__all__ = [ 

165 "AIAuditEvent", 

166 "AIAuditStoreProtocol", 

167 "AIGovernanceProtocol", 

168 "AuditEventType", 

169 "BudgetExceededError", 

170 "CostTrackingProtocol", 

171 "GovernanceDecision", 

172 "GovernanceError", 

173 "PolicyViolationError", 

174 "RelayBillingError", 

175 "RelayBillingProtocol", 

176 "RelayChargeBreakdown", 

177 "RelayPriceEstimatorProtocol", 

178 "RelayUsageRecord", 

179 "RelayUsageReservation", 

180 "RelayUsageScope", 

181 "RelayUsageStoreProtocol", 

182 "ResourceExhaustedError", 

183 "ResourceQuota", 

184 "ResourceUnit", 

185 "ResourceUsageResult", 

186 "ResourceUsageSnapshot", 

187 "ResourceWindowKind", 

188 "billing_store_unavailable", 

189 "charge_overflow", 

190 "duplicate_settlement", 

191 "invalid_usage", 

192 "quota_exhausted", 

193 "reservation_expired", 

194 "unknown_price", 

195]