Guardrails

zeroth.governance.guardrails

Guardrails: operational (rate limiting, quotas, dead-letter) and content safety.

GuardrailConfig

Bases: BaseModel

Tunable guardrail parameters for a deployment.

BlocklistFilter

BlocklistFilter(terms: tuple[str, ...] = ())

Detects (and can redact) configured terms, case-insensitively (literal match).

ContentFilter

Bases: Protocol

A pluggable content filter applied to a single string.

apply

apply(text: str) -> tuple[str, tuple[ContentFinding, ...]]

Return (redacted_text, findings).

The redacted text is only used when the guardrail runs in redact mode; in flag/block mode only the findings matter.

ContentFinding dataclass

ContentFinding(category: str, count: int)

One category of sensitive/disallowed content found in a piece of text.

ContentGuardrail

ContentGuardrail(
    *,
    filters: tuple[ContentFilter, ...]
    | list[ContentFilter],
    mode: str = "flag",
)

Applies a set of ContentFilter s to an agent payload (input or output).

Modes: flag records findings only; redact returns a redacted copy of the payload; block marks the outcome blocked when any finding is present (the caller raises). Walks string leaves of nested dicts/lists.

GuardrailOutcome dataclass

GuardrailOutcome(
    payload: dict,
    findings: tuple[ContentFinding, ...],
    direction: str,
    mode: str,
    blocked: bool,
)

The result of inspecting one payload through a content guardrail.

as_audit

as_audit() -> dict[str, object]

JSON-friendly summary for the audit record.

PIIFilter

PIIFilter(types: tuple[str, ...] = ())

Detects (and can redact) common PII: email, SSN, credit card, phone.

DeadLetterManager dataclass

DeadLetterManager(
    run_repository: RunRepository,
    max_failure_count: int = 3,
)

Escalates a run to dead-letter status after too many failures.

When handle_run_failure is called the failure_count is incremented. If the count reaches max_failure_count the run is transitioned to FAILED with failure_state.reason = "dead_letter" and an admin must explicitly replay it via the admin API.

handle_run_failure async

handle_run_failure(run_id: str) -> bool

Increment failure_count and dead-letter the run if threshold reached.

Returns True if the run was dead-lettered, False otherwise.

QuotaEnforcer

QuotaEnforcer(database: AsyncDatabase)

Per-key rolling-window quota enforcer backed by an async database.

check_and_increment checks whether the counter for a given key is below the configured limit within the current window, and if so atomically increments it. Returns True when within quota, False when exceeded.

table property

table: ScopedTable

Return the structurally scoped table for a coordinated transaction.

check_and_increment async

check_and_increment(
    counter_key: str,
    *,
    limit: int,
    window_seconds: int = 86400,
) -> bool

Check and conditionally increment a quota counter.

Parameters:

Name	Type	Description	Default
counter_key	str	Unique key for the counter (e.g. tenant:daily).	required
limit	int	Maximum allowed increments in the window.	required
window_seconds	int	Duration of the rolling window in seconds.	86400

Returns:

Type	Description
bool	True if within quota (counter incremented), False if exhausted.

decide async

decide(
    counter_key: str,
    *,
    limit: int,
    window_seconds: int = 86400,
) -> QuotaDecision

Increment a rolling quota atomically and return remaining/retry telemetry.

TokenBucketRateLimiter

TokenBucketRateLimiter(database: AsyncDatabase)

Per-key token bucket backed by an async database.

Each bucket has a fixed capacity and refills at a configurable rate. check_and_consume atomically checks whether a token is available and, if so, deducts it. Returns True on success, False when the bucket is empty.

table property

table: ScopedTable

Return the structurally scoped table for a coordinated transaction.

check_and_consume async

check_and_consume(
    bucket_key: str,
    *,
    capacity: float = 10.0,
    refill_rate: float = 1.0,
) -> bool

Attempt to consume one token from the named bucket.

Parameters:

Name	Type	Description	Default
bucket_key	str	Unique key for the bucket (e.g. tenant:deployment).	required
capacity	float	Maximum number of tokens.	10.0
refill_rate	float	Tokens added per second.	1.0

Returns:

Type	Description
bool	True if a token was consumed, False if the bucket is empty.

decide async

decide(
    bucket_key: str,
    *,
    capacity: float = 10.0,
    refill_rate: float = 1.0,
) -> RateLimitDecision

Consume one token atomically and return remaining/retry telemetry.