API Reference

Protocols

AIAuditStore

Protocol for audit event persistence backends.

Implementations must be async and should treat record() as fire-and-forget safe — governance/LLM hot-paths must never block on audit persistence.

async def record(event: AIAuditEvent) -> None

Persist a single audit event.

Parameters

Parameter	Type	Description
`event`	AIAuditEvent	The audit event to store.

async def query(query: AuditQuery) -> list[AIAuditEvent]

Retrieve audit events matching the given filter.

Parameters

Parameter	Type	Description
`query`	AuditQuery	Filter criteria.

Returns

Type	Description
list[AIAuditEvent]	List of matching events ordered by timestamp descending.

async def aggregate(query: AuditQuery) -> AuditSummary

Compute aggregated statistics for matching events.

Parameters

Parameter	Type	Description
`query`	AuditQuery	Filter criteria that scope the aggregation.

Returns

Type	Description
AuditSummary	Summary statistics for the matching events.

---

GovernancePersistence

Storage protocol for governance counters and spend tallies.

Implementations must be safe for concurrent async access. All methods are coroutines to allow either local (in-process) or remote (Redis, database) storage without changing the call-site.

async def incr_requests(
    key: str,
    window: float
) -> int

Record a new request and return the request count within the window.

Parameters

Parameter	Type	Description
`key`	str	Bucket key (e.g. ``"global"`` or a user/tenant id).
`window`	float	Rolling window size in seconds.

Returns

Type	Description
int	Number of requests (including the current one) inside the window.

async def add_spend(
    key: str,
    amount: float,
    ttl: int
) -> float

Add amount to the spend accumulator and return the new total.

Parameters

Parameter	Type	Description
`key`	str	Bucket key (e.g. ``"global:2025-06"``).
`amount`	float	Cost amount to add.
`ttl`	int	Time-to-live for the accumulator entry in seconds.

Returns

Type	Description
float	Updated total spend.

async def get_spend(key: str) -> float

Return the current accumulated spend for key.

Parameters

Parameter	Type	Description
`key`	str	Bucket key.

Returns

Type	Description
float	Current spend; ``0.0`` if no data recorded.

---

Classes

AIAuditEvent

Structured representation of an auditable AI operation.

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

Attributes: event_type: Category of the operation. model: Model identifier (if applicable). provider: Provider name (if applicable). user_id: User who triggered the operation. status: Outcome — "allowed", "denied", "success", "error". tokens: Token count consumed (if applicable). cost: Dollar cost incurred (if applicable). latency_ms: Request latency in milliseconds (if applicable). metadata: Free-form key/value bag for additional context. event_id: Unique UUID for the event (auto-generated).

---

AIGovernanceManager

Enforces AI usage policies: budget limits, rate limits, model restrictions.

Implements AIGovernanceProtocol from contracts.

Governance state (request counts, spend totals) is delegated to a GovernancePersistence backend so the storage strategy is swappable without changing policy logic. When no explicit persistence is given, an InMemoryGovernancePersistence instance is created automatically using the optional cache argument to build a RedisGovernancePersistence when a cache backend is available.

Parameters

Parameter	Type	Description
`config`		Governance policy configuration.
`cache`		Optional cache backend used to auto-create a Redis persistence backend. Ignored when *persistence* is supplied explicitly.
`persistence`		Explicit persistence backend. Takes precedence over *cache*.
`on_soft_limit`		Optional async callback invoked when the monthly spend crosses the ``soft_limit_pct`` threshold. Signature: ``async def cb(user_id, current_spend, budget) -> None``.
`audit_store`		Optional audit store for recording governance decisions. When provided, every governance check (allowed or denied) and every cost-tracking call is recorded as an audit event.

def __init__(
    config: GovernanceConfig,
    cache: CacheBackendProtocol | None = None,
    persistence: GovernancePersistence | None = None,
    on_soft_limit: Callable[Ellipsis, object] | None = None,
    audit_store: AIAuditStore | None = None
) -> None

async def check_request(
    model: str,
    provider: str,
    user_id: str | None = None
) -> bool

Check if a request is allowed under governance policy.

Parameters

Parameter	Type	Description
`model`	str	Model identifier.
`provider`	str	Provider name.
`user_id`	str | None	Optional user identifier for per-user limits.

Returns

Type	Description
bool	True if request is allowed, False if blocked by policy.

def check_model_access(
    user_id: str | None,
    model: str
) -> bool

Check if user is allowed to use the given model.

Evaluates per-user model_allowlist and model_denylist from GovernanceConfig. Both support glob patterns (e.g. "gpt-4*", "claude-3-*").

Logic:

1. If model_allowlist has an entry for user_id, the model must match at least one pattern in the allowlist.
2. If model_denylist has an entry for user_id, the model must not match any pattern in the denylist.
3. When no entry exists for user_id, access is allowed.

Parameters

Parameter	Type	Description
`user_id`	str | None	User identifier, or ``None`` for anonymous / global.
`model`	str	Model name to check.

Returns

Type	Description
bool	True if access is permitted, False if denied.

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

Check if a cost would exceed the monthly budget.

Emits a structured warning when the spend crosses the configured soft_limit_pct threshold and invokes the optional on_soft_limit callback. Returns False only when the hard limit (monthly_budget) would be exceeded.

Parameters

Parameter	Type	Description
`cost`	float	Estimated cost of the request.
`user_id`	str | None	Optional user identifier.

Returns

Type	Description
bool	True if within hard budget, False if would exceed.

async def check_request_budget(
    estimated_cost: float,
    request_id: str | None = None
) -> Result[None, GovernanceError]

Check if a single request cost is within the per-request budget.

Validates estimated_cost against max_request_cost (per-request cap) first, then against the monthly budget via check_budget.

Parameters

Parameter	Type	Description
`estimated_cost`	float	Estimated cost in USD for this request.
`request_id`	str | None	Optional request identifier for logging context.

Returns

Type	Description
Result[None, GovernanceError]	``Ok(None)`` if within all budget limits. ``Err(GovernanceError)`` if either per-request or monthly limit is exceeded.

async def track_cost(
    cost: float,
    model: str,
    user_id: str | None = None
) -> None

Record AI usage cost.

Parameters

Parameter	Type	Description
`cost`	float	Cost to record.
`model`	str	Model that generated the cost.
`user_id`	str | None	Optional user identifier.

def reload_config(config: GovernanceConfig) -> None

Hot-reload governance configuration without restart.

Atomically swaps the internal config reference so that subsequent policy checks use the new limits. Does not touch persistence state — only the thresholds and rules are updated.

Parameters

Parameter	Type	Description
`config`	GovernanceConfig	New governance configuration to apply.

---

AuditEventType

Categories of auditable AI operations.

Shared across lexigram-ai-governance and lexigram-ai-observability via lexigram-contracts.

---

AuditQuery

Filter criteria for querying audit events.

All fields are optional — None means no constraint on that axis.

Attributes: start: Inclusive lower bound on timestamp. end: Inclusive upper bound on timestamp. event_types: Restrict to these event types. user_id: Restrict to a specific user. model: Restrict to a specific model. provider: Restrict to a specific provider. status: Restrict to a specific status string. limit: Maximum number of results to return. offset: Number of results to skip (for pagination).

---

AuditSummary

Aggregated audit statistics for a given query period.

Attributes: total_events: Total number of events matching the query. total_spend: Sum of cost across matching events. total_tokens: Sum of tokens across matching events. denied_count: Events where status is "denied". by_model: Event count per model. by_user: Event count per user. by_event_type: Event count per event type.

---

GovernanceAuditRecordedHook

Payload fired when governance audit logging records an event.

---

GovernanceConfig

Configuration for AI Governance.

Loaded from the ai_governance: key in application.yaml, with environment variable overrides via LEX_AI_GOVERNANCE__* prefix.

def validate_for_environment(env: Environment | None = None) -> list[ConfigIssue]

Check config is safe for the target environment.

---

GovernanceModule

AI Governance policy enforcement and cost-tracking integration.

Call configure to register AIGovernanceProtocol and CostTrackingProtocol implementations for injection.

Usage

from lexigram.ai.governance.config import GovernanceConfig

@module(
    imports=[
        GovernanceModule.configure(
            GovernanceConfig(monthly_budget=100.0)
        )
    ]
)
class AppModule(Module):
    pass

Error Handling

Governance violations surface as typed exceptions that can be caught
directly or handled via the Result pattern::

    from lexigram.ai.governance.exceptions import (
        GovernanceError,       # base — catch-all
        BudgetExceededError,   # monthly spend cap breached
        RateLimitExceededError,# RPM / TPM limit exceeded
        ModelAccessDeniedError,# policy denied model access
    )

Exports: AIGovernanceProtocol, CostTrackingProtocol, GovernanceError, BudgetExceededError, RateLimitExceededError, ModelAccessDeniedError

def configure(
    cls,
    config: GovernanceConfig | None = None
) -> DynamicModule

Create a GovernanceModule with the given configuration.

Parameters

Parameter	Type	Description
`config`	GovernanceConfig | None	GovernanceConfig, a plain ``dict`` of the same keys, or ``None`` to read from environment variables.

Returns

Type	Description
DynamicModule	A DynamicModule descriptor.

Raises

Exception	Description
TypeError	If *config* is not a ``GovernanceConfig``, ``dict``, or ``None``.

def stub(
    cls,
    config: GovernanceConfig | None = None
) -> DynamicModule

Create a GovernanceModule suitable for unit and integration testing.

Uses in-memory or no-op implementations with minimal side effects.

Parameters

Parameter	Type	Description
`config`	GovernanceConfig | None	Optional config override. Uses safe test defaults when None.

Returns

Type	Description
DynamicModule	A DynamicModule descriptor.

---

GovernancePersistenceWrittenHook

Payload fired when governance persistence writes a record.

---

GovernancePolicyEvaluatedHook

Payload fired when the governance policy layer evaluates a policy.

---

GovernanceProvider

Provider for AI Governance.

Registers AIGovernanceManager.

def __init__(config: GovernanceConfig | dict | None = None) -> None

def from_config(
    cls,
    config: GovernanceConfig,
    **context: object
) -> GovernanceProvider

Factory method for DI container setup.

async def register(container: ContainerRegistrarProtocol) -> None

Register the governance services.

async def boot(container: ContainerResolverProtocol) -> None

Boot phase.

async def shutdown() -> None

Shutdown phase.

async def health_check(timeout: float = 5.0) -> HealthCheckResult

Health check — always healthy (in-process domain provider).

No external backend to ping.

Parameters

Parameter	Type	Description
`timeout`	float	Ignored for in-process providers.

Returns

Type	Description
HealthCheckResult	Always HEALTHY — no external backend to ping.

---

InMemoryAuditStore

In-memory audit store for testing and development.

Not suitable for production — events are lost on process restart. The store is intentionally simple: append-only list with linear scan queries. For production use, implement AIAuditStore with a durable backend (e.g. via DatabaseProviderProtocol).

def __init__() -> None

async def record(event: AIAuditEvent) -> None

Append event to the in-memory list.

async def query(query: AuditQuery) -> list[AIAuditEvent]

Linear scan with filter, ordered by timestamp descending.

async def aggregate(query: AuditQuery) -> AuditSummary

Compute summary statistics over matching events.

---

InMemoryGovernancePersistence

Process-local governance persistence using plain Python dicts.

Request counts are tracked with a sliding-window approach (list of monotonic timestamps). Spend totals are stored as plain floats.

This implementation is not suitable for multi-process or multi-replica deployments. Use RedisGovernancePersistence in production.

def __init__() -> None

async def incr_requests(
    key: str,
    window: float
) -> int

async def add_spend(
    key: str,
    amount: float,
    ttl: int
) -> float

async def get_spend(key: str) -> float

---

PolicyEvaluatedEvent

Emitted when a governance policy evaluation completes.

Consumed by: audit, compliance, cost management.

---

RedisGovernancePersistence

Distributed governance persistence backed by a Lexigram CacheBackendProtocol.

Request windows use a sorted-set approach:

• Each request is stored as a member with its Unix timestamp as score.
• Expired members (score < now - window) are pruned on every read.

Spend totals are stored as plain string floats with a configurable TTL so that monthly counters expire automatically.

Parameters

Parameter	Type	Description
`cache`		A CacheBackendProtocol that has been connected and is ready to accept commands. The backend is expected to be Redis-compatible.

def __init__(cache: CacheBackendProtocol) -> None

async def incr_requests(
    key: str,
    window: float
) -> int

Use a sorted set to implement a sliding window counter.

Falls back to an approximate counter if the backend does not support sorted-set operations (e.g. a simple in-memory mock).

async def add_spend(
    key: str,
    amount: float,
    ttl: int
) -> float

async def get_spend(key: str) -> float

---

Exceptions

BudgetExceededError

Error raised when budget is exceeded.

---

GovernanceError

Base class for governance-related errors.

---

ModelAccessDeniedError

Raised when a user is denied access to a model by policy.

Attributes: model: Model identifier that was denied. reason: Why access was denied ("restricted", "not_in_allowlist", "in_denylist").

def __init__(
    model: str,
    reason: str,
    user_id: str | None = None
) -> None

---

RateLimitExceededError

Raised when RPM or TPM limits are exceeded.

Attributes: limit: Configured limit value. current: Current counter value. limit_type: Type of limit exceeded ("rpm" or "tpm").

def __init__(
    limit: int,
    current: int,
    limit_type: str = 'rpm',
    user_id: str | None = None
) -> None

---