Coverage for /home/admin/Documents/AI/applications/lexigram-dev/lexigram/experimental/ai/lexigram-ai-governance/src/lexigram/ai/governance/config.py: 79%

33 statements  

« prev     ^ index     » next       coverage.py v7.15.4, created at 2026-08-25 07:19 +0800

1"""Configuration for AI Governance.""" 

2 

3from __future__ import annotations 

4 

5from dataclasses import dataclass 

6from typing import TYPE_CHECKING, ClassVar, cast 

7 

8from lexigram.ai.governance import constants as const 

9from lexigram.config import ( 

10 BaseConfig, 

11) 

12from lexigram.contracts.core.config import ConfigIssue, Environment 

13from lexigram.validation import ConfigDict, Field 

14 

15if TYPE_CHECKING: 

16 from lexigram.contracts.ai.governance.resource_unit import ( # noqa: F401 

17 ResourceUnit, 

18 ) 

19 

20 

21@dataclass(init=False) 

22class GovernanceConfig(BaseConfig): 

23 """Configuration for AI Governance. 

24 

25 Loaded from the ``ai_governance:`` key in application.yaml, with environment 

26 variable overrides via ``LEX_AI_GOVERNANCE__*`` prefix. 

27 """ 

28 

29 config_section: ClassVar[str] = "ai_governance" 

30 

31 model_config: ClassVar[ConfigDict] = cast( 

32 "ConfigDict", 

33 { 

34 "env_prefix": const.ENV_PREFIX, 

35 "env_nested_delimiter": const.ENV_NESTED_DELIMITER, 

36 "extra": "ignore", 

37 }, 

38 ) 

39 

40 enabled: bool = Field(default=True, description="Enable AI governance") 

41 monthly_budget: float | None = Field( 

42 default=None, 

43 description="Monthly budget in dollars", 

44 ) 

45 max_tokens_per_request: int | None = Field( 

46 default=None, 

47 description="Max tokens per request", 

48 ) 

49 rpm_limit: int | None = Field(default=None, description="Requests Per Minute limit") 

50 tpm_limit: int | None = Field(default=None, description="Tokens Per Minute limit") 

51 restricted_models: list[str] = Field( 

52 default_factory=list, 

53 description="List of restricted models", 

54 ) 

55 enforce_budget: bool = Field(default=True, description="Enforce budget limits") 

56 soft_limit_pct: float | None = Field( 

57 default=None, 

58 description=( 

59 "Fraction of monthly_budget at which to emit a soft-limit warning " 

60 "(e.g. 0.8 = warn at 80%). No hard block is applied at this threshold." 

61 ), 

62 ge=0.0, 

63 le=1.0, 

64 ) 

65 max_request_cost: float | None = Field( 

66 default=None, 

67 description=( 

68 "Maximum cost in dollars for a single request. " 

69 "Requests with an estimated cost above this threshold are rejected " 

70 "before they reach the monthly-budget check." 

71 ), 

72 ge=0.0, 

73 ) 

74 model_allowlist: dict[str, list[str]] = Field( 

75 default_factory=dict, 

76 description=( 

77 "Per-user/role model allowlist. Keys are user IDs or role names; " 

78 "values are lists of allowed model patterns (supports glob syntax, " 

79 "e.g. 'gpt-4*', 'claude-3-*')." 

80 ), 

81 ) 

82 model_denylist: dict[str, list[str]] = Field( 

83 default_factory=dict, 

84 description=( 

85 "Per-user/role model denylist. Keys are user IDs or role names; " 

86 "values are lists of denied model patterns (supports glob syntax). " 

87 "Denylist is checked after allowlist." 

88 ), 

89 ) 

90 resource_units: list = Field( 

91 default_factory=list, 

92 description=( 

93 "Resource units this governance instance tracks. " 

94 "Per-tenant limits are configured via TenantConfigService overrides." 

95 ), 

96 ) 

97 fail_open_on_persistence_error: bool = Field( 

98 default=False, 

99 description=( 

100 "Allow requests when the persistence backend is unavailable. " 

101 "When False (default, fail-closed), a persistence failure (e.g. Redis " 

102 "down) denies the request: the budget check treats spend as unknown " 

103 "and denies, the RPM check denies, and cost recording is skipped; " 

104 "every failure is logged. When True (fail-open), the same failure " 

105 "allows the request and skips cost recording, trading spend/rate " 

106 "enforcement for availability during infrastructure outages." 

107 ), 

108 ) 

109 

110 def validate_for_environment( 

111 self, env: Environment | None = None 

112 ) -> list[ConfigIssue]: 

113 """Check config is safe for the target environment.""" 

114 issues: list[ConfigIssue] = [] 

115 

116 if env == Environment.PRODUCTION: 

117 if not self.enforce_budget: 

118 issues.append( 

119 ConfigIssue( 

120 severity="warning", 

121 field="enforce_budget", 

122 message="Budget enforcement disabled in production", 

123 suggestion=( 

124 f"Set {const.ENV_PREFIX}ENFORCE_BUDGET=true or remove " 

125 "the override to enforce budget limits." 

126 ), 

127 ) 

128 ) 

129 if self.monthly_budget is None: 

130 issues.append( 

131 ConfigIssue( 

132 severity="warning", 

133 field="monthly_budget", 

134 message="No monthly budget configured in production", 

135 suggestion=( 

136 f"Set {const.ENV_PREFIX}MONTHLY_BUDGET to a dollar amount " 

137 "to prevent runaway costs." 

138 ), 

139 ) 

140 ) 

141 

142 return issues 

143 

144 

145__all__ = ["GovernanceConfig"]