Skip to content

API: techrevati.runtime.governance

techrevati.runtime.governance

Governance plane — hard-stop limits enforced outside agent code.

The governance plane is the runtime's last line of defense against runaway agent loops, budget overruns, persistent failures, and unbounded tool use. Unlike UsageLimits (which the agent can in principle catch + recover from), a GovernanceBreachError is terminal: it propagates through the orchestrator without entering the recovery loop or the failure classifier.

This is the Waxell pattern: enforcement at the governance plane, not inside agent logic, so the agent cannot bypass its own limits. For EU AI Act Article 14 (human oversight) and Article 15 (robustness) deployments this is the technical primitive auditors expect to see.

Each limit is a small frozen dataclass with three pieces of data:

  • value — the ceiling.
  • scope — currently only "session" is enforced; "thread" and "project" are reserved for future cross-session enforcement.
  • on_breach — "terminate" raises GovernanceBreachError; "alert" emits a governance.alert event and continues. Use "alert" during rollout to measure breach rates before flipping to "terminate" in production.

Composition into a session:

from techrevati.runtime import AgentSession from techrevati.runtime.governance import ( ... GovernancePlane, MaxIterationsLimit, MaxBudgetLimit, ... ) plane = GovernancePlane( ... limits=( ... MaxIterationsLimit(value=10), ... MaxBudgetLimit(value=5.00, on_breach="alert"), ... ), ... ) session = AgentSession(role="writer", phase="draft", governance=plane)

GovernanceBreachError 

GovernanceBreachError(*, limit_name, observed, ceiling, scope)

Bases: Exception

Raised when a governance limit is exceeded with on_breach="terminate".

Terminal by contract: the orchestrator re-raises this without invoking classify_exception or attempt_recovery. Caller code that wraps a session in except Exception will still see it; if that's a problem in your code path, catch GovernanceBreachError explicitly before the broader handler.

GovernanceState dataclass 

GovernanceState(turns=0, tool_calls=0, consecutive_failures=0, cost_usd=0.0)

Mutable accumulator the plane reads on each evaluation.

The orchestrator updates these counters as the session progresses; the plane never mutates them. Keep additions on this state cheap — it is read on every turn and every tool call.

MaxIterationsLimit dataclass 

MaxIterationsLimit(value, scope='session', on_breach='terminate', name='max_iterations')

Bases: _LimitBase

Cap total turns in the session. Pairs with AgentSession.max_iterations.

Difference from AgentSession.max_iterations: the latter raises a domain MaxIterationsExceededError which can be caught and handled inside agent code. MaxIterationsLimit with on_breach="terminate" raises GovernanceBreachError which is terminal and skips the recovery loop.

MaxBudgetLimit dataclass 

MaxBudgetLimit(value, scope='session', on_breach='terminate', name='max_budget_usd')

Bases: _LimitBase

Cap cumulative cost in USD. Pairs with UsageLimits.cost_usd_max.

Same difference as MaxIterationsLimit: this is the hard-stop twin of the recoverable BudgetExceededError.

MaxConsecutiveFailuresLimit dataclass 

MaxConsecutiveFailuresLimit(value, scope='session', on_breach='terminate', name='max_consecutive_failures')

Bases: _LimitBase

Cap consecutive recovery failures.

Counts only consecutive failures — a single success resets the counter. Catches "agent retries the same broken thing forever" failure modes that RecoveryRecipe step retries alone do not.

MaxToolCallsLimit dataclass 

MaxToolCallsLimit(value, scope='session', on_breach='terminate', name='max_tool_calls')

Bases: _LimitBase

Cap total tool calls in the session.

Distinct from UsageLimits.tool_calls_max only in that this is a hard-stop. Choose this when "the agent is spamming tool calls" is a safety / cost concern, not a usage telemetry concern.

LimitOutcome dataclass 

LimitOutcome(breached, limit_name, observed, ceiling, scope, on_breach)

Result of evaluating one limit against current state.

GovernancePlane dataclass 

GovernancePlane(limits, state=GovernanceState())

Composes a set of limits and enforces them against a GovernanceState.

The orchestrator constructs and owns the state; the plane is a pure evaluator + raiser. This separation keeps the limit objects immutable and the per-session counter mutable, which is what tests and audit replays both want.

evaluate 

evaluate()

Evaluate every limit. Returns outcomes (breached or not).

Does NOT raise. Callers that want enforcement call enforce() instead; this method is exposed for inspection and for the "alert" code path which only emits events.

enforce 

enforce()

Evaluate and raise on the first on_breach="terminate" breach.

Returns the full list of outcomes (including breached ones with on_breach="alert") so the caller can record them via the event sink. Raises GovernanceBreachError on the first terminate-breach.