API Reference

Protocols

AIAuditStore

source

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.

record

async def record(event: AIAuditEvent) -> None

source

Persist a single audit event.

Parameters

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

query

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

source

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.

aggregate

async def aggregate(query: AuditQuery) -> AuditSummary

source

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.

---

AIAuditStoreProtocol

source

Protocol for AI audit event persistence backends.

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

record

async def record(event: AIAuditEvent) -> None

source

Persist a single audit event.

Parameters

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

---

AIGovernanceProtocol

source

Protocol for AI Governance and cost tracking.

check_request

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

source

Check if request is allowed by governance policies.

check_budget

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

source

Check if operation fits within budget.

track_cost

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

source

Track cost for an operation.

---

ContentPolicyGateProtocol

source

Gate that decides whether a request should reach the LLM.

Implementations are responsible for a single evaluation concern (e.g. abuse detection, quota enforcement, age gating, jurisdictional restrictions). Complex policies are composed via CompositeGate.

The method must be async and must never raise for domain-level outcomes — DENY is returned as a PolicyDecision, not raised as an exception. Infrastructure failures (DB down, etc.) may raise.

evaluate

async def evaluate(request: PolicyRequest) -> PolicyDecision

source

Evaluate request and return a policy decision.

Parameters

Parameter	Type	Description
`request`	PolicyRequest	Contextual information about the content and caller.

Returns

Type	Description
PolicyDecision	A PolicyDecision with one of ``ALLOW``, ``DENY``, or ``THROTTLE``.

---

CostTrackingProtocol

source

Protocol for tracking costs.

track_cost

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

source

Track cost for an operation.

get_budget

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

source

Get remaining budget.

---

GaugeReconciliationCallback

source

Source-of-truth interface for an INSTANTANEOUS gauge.

Implementations are domain-specific — e.g. for concurrent_episodes_in_production_per_tenant, count_active queries the saga store for RUNNING episode-production sagas owned by tenant_id.

list_tenants

async def list_tenants() -> list[str]

source

Return all tenant IDs to reconcile this cycle.

count_active

async def count_active(tenant_id: str) -> float

source

Return the ground-truth held count for tenant_id.

---

GovernancePersistence

source

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.

incr_requests

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

source

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.

add_spend

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

source

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.

get_spend

async def get_spend(key: str) -> float

source

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.

read_gauge

async def read_gauge(key: str) -> float

source

Read the current gauge value for key.

Parameters

Parameter	Type	Description
`key`	str	Gauge key (e.g. ``"tenant:gpt4:remaining"``).

Returns

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

write_gauge

async def write_gauge(
    key: str,
    value: float,
    ttl: int
) -> None

source

Set the gauge value for key with a TTL.

Parameters

Parameter	Type	Description
`key`	str	Gauge key.
`value`	float	Float gauge value.
`ttl`	int	Time-to-live in seconds.

incr_gauge

async def incr_gauge(
    key: str,
    delta: float,
    ttl: int
) -> float

source

Atomically increment (or decrement) the gauge for key.

Parameters

Parameter	Type	Description
`key`	str	Gauge key.
`delta`	float	Amount to add (can be negative).
`ttl`	int	Time-to-live in seconds.

Returns

Type	Description
float	The gauge value after applying *delta*.

add_calendar_entry

async def add_calendar_entry(
    key: str,
    timestamp: float,
    ttl: int
) -> None

source

Record a timestamp entry in a calendar-style bucket.

Parameters

Parameter	Type	Description
`key`	str	Calendar bucket key.
`timestamp`	float	Unix timestamp to record.
`ttl`	int	Time-to-live in seconds for the bucket.

query_calendar

async def query_calendar(
    key: str,
    start: float,
    end: float
) -> list[float]

source

Return all timestamps in key that fall within [start, end].

Parameters

Parameter	Type	Description
`key`	str	Calendar bucket key.
`start`	float	Unix timestamp, start of range (inclusive).
`end`	float	Unix timestamp, end of range (inclusive).

Returns

Type	Description
list[float]	List of matching timestamps (chronological order).

decr_gauge

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

source

Decrement a gauge key by amount.

Default implementation delegates to incr_gauge with a negative delta.

incr_calendar

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

source

Accumulate amount in a calendar-window bucket.

Delegates to add_spend with a period-scoped key (key + : + period).

get_calendar

async def get_calendar(
    key: str,
    period: str
) -> float

source

Read the calendar-window accumulator for key + period.

Delegates to get_spend with a period-scoped key.

---

PolicyObserverProtocol

source

Side-effect hook called after every gate evaluation.

Observers receive the original request and the final decision. They must not mutate the decision. Use for metrics, external audit streams, or custom structured logging.

Observers are invoked asynchronously by CompositeGate and ContentPolicyService. Infrastructure errors inside an observer must be handled by the observer implementation.

on_decision

async def on_decision(
    request: PolicyRequest,
    decision: PolicyDecision
) -> None

source

Called after a gate produces a decision.

Parameters

Parameter	Type	Description
`request`	PolicyRequest	The original policy request.
`decision`	PolicyDecision	The decision that was produced.

---

Classes

AIAuditEvent

source

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

source

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.

__init__

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

source

resource_tracker

property resource_tracker() -> ResourceUnitTracker | None

source

The ResourceUnitTracker instance, or None when no resource units are configured.

Exposed for DI registration so the same tracker is shared across the application (consume/release calls go through here regardless of whether the caller resolves AIGovernanceManager or ResourceUnitTracker from the container).

check_request

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

source

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.

check_model_access

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

source

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.

check_budget

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

source

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.

check_request_budget

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

source

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.

track_cost

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

source

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.

reload_config

def reload_config(config: GovernanceConfig) -> None

source

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.

consume_resource

async def consume_resource(
    tenant_id: str,
    unit_name: str,
    amount: float,
    actor_id: str | None = None
) -> Result

source

Consume amount of a resource unit for tenant_id.

Delegates to ResourceUnitTracker if configured.

release_resource

async def release_resource(
    tenant_id: str,
    unit_name: str,
    amount: float
) -> None

source

Release amount of a held resource (INSTANTANEOUS units only).

resource_usage

async def resource_usage(
    tenant_id: str,
    unit_name: str
)

source

Return current usage snapshot for tenant_id + unit_name.

---

AlwaysAllowGate

source

Trivial gate that allows every request.

Use as a default/no-op gate in development, testing, or when the application does not yet have a content policy.

evaluate

async def evaluate(request: PolicyRequest) -> PolicyDecision

source

Always return ALLOW.

Parameters

Parameter	Type	Description
`request`	PolicyRequest	Policy request (ignored).

Returns

Type	Description
PolicyDecision	``PolicyDecision(outcome=ALLOW, reason="")``.

---

AuditEventType

source

Categories of auditable AI operations.

Shared across the AI governance and observability layers via lexigram-contracts.

---

AuditQuery

source

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

source

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.

---

CompositeGate

source

Combines multiple ContentPolicyGateProtocol gates.

Evaluation algorithm:

1. All member gates are evaluated (no short-circuit on DENY so that all observers receive the full picture).
2. If any gate returns DENY → the composite returns DENY.
3. Otherwise the most-restrictive outcome across all gates wins (THROTTLE beats ALLOW).
4. All non-empty reason strings from member gates are aggregated, separated by "; ".
5. After the final decision is produced, each registered PolicyObserverProtocol is notified.

Parameters

Parameter	Type	Description
`gates`		One or more gate instances to compose.
`observers`		Optional observers notified after each composite evaluation.

__init__

def __init__(
    gates: Sequence[ContentPolicyGateProtocol],
    observers: Sequence[PolicyObserverProtocol] | None = None
) -> None

source

evaluate

async def evaluate(request: PolicyRequest) -> PolicyDecision

source

Evaluate request against all member gates and return a composite decision.

Parameters

Parameter	Type	Description
`request`	PolicyRequest	Contextual information about the content and caller.

Returns

Type	Description
PolicyDecision	The most-restrictive PolicyDecision across all member gates, with aggregated reasons.

---

ContentPolicyService

source

Evaluates content policy before an LLM request is dispatched.

This service is the primary DI-injectable surface for content gating. It wraps a single gate (which may itself be a CompositeGate) and adds structured logging for every decision so that operators have an audit trail without coupling gate implementations to a specific logger.

Parameters

Parameter	Type	Description
`gate`		The gate (or composite) to delegate to.
`observers`		Additional observers notified after each evaluation. These are layered on top of any observers already configured in the gate itself.

__init__

def __init__(
    gate: ContentPolicyGateProtocol,
    observers: list[PolicyObserverProtocol] | None = None
) -> None

source

evaluate

async def evaluate(request: PolicyRequest) -> PolicyDecision

source

Evaluate request and emit a structured log event.

Apps call this method before dispatching to an LLM. If the returned outcome is DENY, the caller must not proceed

decision = await policy_service.evaluate(request)
if decision.outcome == PolicyOutcome.DENY:
    raise ContentPolicyViolation(decision.reason)

decision = await policy_service.evaluate(request)
if decision.outcome == PolicyOutcome.DENY:
    raise ContentPolicyViolation(decision.reason)

Parameters

Parameter	Type	Description
`request`	PolicyRequest	Context about the content and the requesting principal.

Returns

Type	Description
PolicyDecision	A PolicyDecision. Never raises for domain-level outcomes; infrastructure errors propagate normally.

is_allowed

def is_allowed(decision: PolicyDecision) -> bool

source

Convenience predicate — True for ALLOW or THROTTLE outcomes.

Parameters

Parameter	Type	Description
`decision`	PolicyDecision	A decision previously returned by evaluate.

Returns

Type	Description
bool	``False`` only when *outcome* is ``DENY``.

---

GaugeReconciliationWorker

source

Periodically reconciles INSTANTANEOUS gauges against ground truth.

Each cycle: for every (unit_name, callback) registered, list the tenants and call count_active to obtain the correct value; write it via ResourceUnitTracker.reconcile. Per-tenant errors are logged and skipped — a single bad callback never poisons the cycle.

Default cadence is 5 minutes; tunable per-instance.

__init__

def __init__(
    task_manager: BackgroundTaskManager,
    tracker: ResourceUnitTracker,
    callbacks: dict[str, GaugeReconciliationCallback] | None = None,
    *,
    interval_seconds: float | None = None
) -> None

source

register

def register(
    unit_name: str,
    callback: GaugeReconciliationCallback
) -> None

source

Register a reconciliation callback for unit_name.

unregister

def unregister(unit_name: str) -> None

source

Remove the callback for unit_name (no-op if absent).

callbacks

property callbacks() -> dict[str, GaugeReconciliationCallback]

source

Snapshot of currently-registered callbacks.

run_cycle

async def run_cycle() -> None

source

---

GovernanceAuditRecordedHook

source

Payload fired when governance audit logging records an event.

---

GovernanceConfig

source

Configuration for AI Governance.

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

validate_for_environment

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

source

Check config is safe for the target environment.

---

GovernanceModule

source

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

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
    )

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
)

    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

configure

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

source

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``.

stub

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

source

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

source

Payload fired when governance persistence writes a record.

---

GovernancePolicyEvaluatedHook

source

Payload fired when the governance policy layer evaluates a policy.

---

GovernanceProvider

source

Provider for AI Governance.

Registers AIGovernanceManager.

__init__

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

source

from_config

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

source

Factory method for DI container setup.

register

async def register(container: ContainerRegistrarProtocol) -> None

source

Register the governance services.

boot

async def boot(container: ContainerResolverProtocol) -> None

source

Boot phase.

shutdown

async def shutdown() -> None

source

Shutdown phase.

health_check

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

source

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

source

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).

__init__

def __init__() -> None

source

record

async def record(event: AIAuditEvent) -> None

source

Append event to the in-memory list.

query

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

source

Linear scan with filter, ordered by timestamp descending.

aggregate

async def aggregate(query: AuditQuery) -> AuditSummary

source

Compute summary statistics over matching events.

---

InMemoryGovernancePersistence

source

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.

__init__

def __init__() -> None

source

incr_requests

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

source

add_spend

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

source

get_spend

async def get_spend(key: str) -> float

source

read_gauge

async def read_gauge(key: str) -> float

source

write_gauge

async def write_gauge(
    key: str,
    value: float,
    ttl: int
) -> None

source

incr_gauge

async def incr_gauge(
    key: str,
    delta: float,
    ttl: int
) -> float

source

add_calendar_entry

async def add_calendar_entry(
    key: str,
    timestamp: float,
    ttl: int
) -> None

source

query_calendar

async def query_calendar(
    key: str,
    start: float,
    end: float
) -> list[float]

source

---

PolicyDecision

source

Result produced by a ContentPolicyGateProtocol.

Decisions are values, not exceptions — even a DENY is returned normally. The caller decides what to do.

Attributes: outcome: One of ALLOW, DENY, or THROTTLE. reason: Human-readable explanation (shown in logs / returned to the app; may be empty for ALLOW). metadata: Additional context produced by the gate (e.g. {"threshold": "5", "current_count": "6"}).

---

PolicyEvaluatedEvent

source

Emitted when a governance policy evaluation completes.

Consumed by: audit, compliance, cost management.

---

PolicyOutcome

source

Result of a content policy gate evaluation.

• ALLOW — request is permitted.
• DENY — request is blocked; the caller must not proceed.
• THROTTLE — allow now but record; future requests may be denied.

---

PolicyRequest

source

Context passed to a ContentPolicyGateProtocol.

Attributes: principal_id: Identifies the caller (user-id, session token, etc.). None for fully anonymous requests. content_type: Application-defined label for what is being submitted (e.g. "image", "text", "document"). content_metadata: Free-form key/value pairs describing the content (e.g. {"species": "cat", "mime_type": "image/jpeg"}). history_summary: Aggregated counters from prior requests by this principal (e.g. {"non_pet_images": 3}). None when no history is available.

---

RedisGovernancePersistence

source

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.

__init__

def __init__(cache: CacheBackendProtocol) -> None

source

incr_requests

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

source

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).

add_spend

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

source

get_spend

async def get_spend(key: str) -> float

source

read_gauge

async def read_gauge(key: str) -> float

source

write_gauge

async def write_gauge(
    key: str,
    value: float,
    ttl: int
) -> None

source

incr_gauge

async def incr_gauge(
    key: str,
    delta: float,
    ttl: int
) -> float

source

add_calendar_entry

async def add_calendar_entry(
    key: str,
    timestamp: float,
    ttl: int
) -> None

source

query_calendar

async def query_calendar(
    key: str,
    start: float,
    end: float
) -> list[float]

source

---

ResourceUnitRegistry

source

In-memory registry of known ResourceUnit definitions.

Populated at boot from GovernanceConfig.resource_units via the DI provider. Lookup is by unit name (ResourceUnit.name).

__init__

def __init__() -> None

source

register

def register(unit: ResourceUnit) -> None

source

Register (or overwrite) a resource unit definition.

get_unit

def get_unit(name: str) -> ResourceUnit | None

source

Look up a resource unit by name, or None.

list_units

def list_units() -> list[ResourceUnit]

source

Return all registered resource units.

from_list

def from_list(
    cls,
    units: list[ResourceUnit]
) -> ResourceUnitRegistry

source

Create a registry pre-populated with units.

---

ResourceUnitTracker

source

Tracks per-tenant resource consumption via the persistence layer.

Routes consume / release / usage calls to the right persistence method based on the unit’s window_kind:

• SLIDING → incr_requests (existing sliding-window counter)
• CALENDAR → incr_calendar / get_calendar
• INSTANTANEOUS → incr_gauge / decr_gauge / read_gauge

Parameters

Parameter	Type	Description
`registry`		Registry of known ResourceUnit definitions.
`persistence`		Backend that implements GovernancePersistence.
`get_quota`		Callable ``(tenant_id, unit_name) → limit``. Return ``0.0`` to deny all usage.

__init__

def __init__(
    registry: ResourceUnitRegistry,
    persistence: GovernancePersistence,
    get_quota: Callable[[str, str], float] | None = None
) -> None

source

consume

async def consume(
    tenant_id: str,
    unit_name: str,
    amount: float,
    actor_id: str | None = None
) -> Result[ResourceUsageSnapshot, ResourceExhaustedError]

source

Consume amount of a resource for tenant_id.

Returns

Type	Description
Result[ResourceUsageSnapshot, ResourceExhaustedError]	``Ok(snapshot)`` on success, ``Err(ResourceExhaustedError)`` if the quota would be exceeded.

release

async def release(
    tenant_id: str,
    unit_name: str,
    amount: float
) -> None

source

Release amount of a held resource (INSTANTANEOUS units only).

For SLIDING and CALENDAR windows this is a no-op — windows decay naturally.

reconcile

async def reconcile(
    tenant_id: str,
    unit_name: str,
    expected_amount: float
) -> None

source

Reset the gauge for an INSTANTANEOUS unit to expected_amount.

Used by GaugeReconciliationWorker to repair drift caused by crashed processes that consumed without releasing. No-op for non-INSTANTANEOUS units.

usage

async def usage(
    tenant_id: str,
    unit_name: str
) -> ResourceUsageSnapshot

source

Return current usage snapshot for tenant_id + unit_name.

---

Exceptions

BudgetExceededError

source

Error raised when budget is exceeded.

---

GovernanceError

source

Base class for governance-related errors.

---

ModelAccessDeniedError

source

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").

__init__

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

source

---

RateLimitExceededError

source

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").

__init__

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

source

---

ResourceExhaustedError

source

Raised when a resource quota is exhausted.

__init__

def __init__(
    tenant_id: str,
    unit_name: str,
    limit: float,
    current: float,
    actor_id: str | None = None
) -> None

source

---