agenticraft_foundation.verification.invariant_checker¶

Runtime invariant checker — register invariants, check state transitions, track violations.

Runtime invariant checking for distributed systems.

This module provides tools for runtime verification of distributed system invariants, enabling early detection of protocol violations.

Features: - State transition monitoring - Invariant assertion and logging - Violation reporting and debugging - Thread-safe concurrent access via RLock

Thread Safety

All mutable state in InvariantRegistry is protected by RLock, enabling safe concurrent invariant checking across threads.

Version: 0.1.0 Date: 2025-12-20

`ViolationSeverity` ¶

Bases: str, Enum

Severity levels for invariant violations.

`Invariant` `dataclass` ¶

A runtime invariant to be checked.

ATTRIBUTE	DESCRIPTION
`name`	Descriptive name TYPE: `str`
`condition`	Function that returns True if invariant holds TYPE: `Callable[..., bool]`
`severity`	Severity level if violated TYPE: `ViolationSeverity`
`message`	Message to log on violation TYPE: `str`
`enabled`	Whether invariant checking is enabled TYPE: `bool`

`check(*args, **kwargs)` ¶

Check if invariant holds.

PARAMETER	DESCRIPTION
`*args`	Arguments to pass to condition TYPE: `Any` DEFAULT: `()`
`**kwargs`	Keyword arguments to pass to condition TYPE: `Any` DEFAULT: `{}`

RETURNS	DESCRIPTION
`bool`	True if invariant holds

`Violation` `dataclass` ¶

Record of an invariant violation.

ATTRIBUTE	DESCRIPTION
`invariant_name`	Name of violated invariant TYPE: `str`
`severity`	Severity of violation TYPE: `ViolationSeverity`
`message`	Violation message TYPE: `str`
`context`	Context at time of violation TYPE: `dict[str, Any]`
`timestamp`	When violation occurred TYPE: `float`
`stack_trace`	Optional stack trace TYPE: `str \| None`

`InvariantRegistry` ¶

Registry for managing system invariants.

Provides centralized management of invariants with checking and violation tracking.

Thread Safety

All mutable state is protected by RLock, enabling safe concurrent access from multiple threads. RLock is used (vs Lock) to allow re-entrant calls from violation handlers.

`init(name='default')` ¶

Initialize registry.

PARAMETER	DESCRIPTION
`name`	Registry name for logging TYPE: `str` DEFAULT: `'default'`

`register(name, condition, severity=ViolationSeverity.ERROR, message='')` ¶

Register a new invariant.

Thread-safe: Protected by RLock.

PARAMETER	DESCRIPTION
`name`	Unique name for invariant TYPE: `str`
`condition`	Function returning True if invariant holds TYPE: `Callable[..., bool]`
`severity`	Severity level if violated TYPE: `ViolationSeverity` DEFAULT: `ERROR`
`message`	Message to log on violation TYPE: `str` DEFAULT: `''`

RETURNS	DESCRIPTION
`Invariant`	The registered Invariant

`unregister(name)` ¶

Unregister an invariant.

Thread-safe: Protected by RLock.

PARAMETER	DESCRIPTION
`name`	Invariant name to remove TYPE: `str`

RETURNS	DESCRIPTION
`bool`	True if invariant was removed

`check(name, *args, context=None, **kwargs)` ¶

Check a specific invariant.

Thread-safe: Protected by RLock. The invariant condition is evaluated outside the lock to avoid blocking concurrent access during potentially long-running checks.

PARAMETER	DESCRIPTION
`name`	Invariant name to check TYPE: `str`
`*args`	Arguments to pass to invariant condition TYPE: `Any` DEFAULT: `()`
`context`	Additional context for violation reporting TYPE: `dict[str, Any] \| None` DEFAULT: `None`
`**kwargs`	Keyword arguments to pass to condition TYPE: `Any` DEFAULT: `{}`

RETURNS	DESCRIPTION
`bool`	True if invariant holds

RAISES	DESCRIPTION
`KeyError`	If invariant not found

`check_all(*args, context=None, **kwargs)` ¶

Check all registered invariants.

Thread-safe: Iterates over a snapshot of invariant names.

PARAMETER	DESCRIPTION
`*args`	Arguments to pass to invariant conditions TYPE: `Any` DEFAULT: `()`
`context`	Additional context for violation reporting TYPE: `dict[str, Any] \| None` DEFAULT: `None`
`**kwargs`	Keyword arguments to pass to conditions TYPE: `Any` DEFAULT: `{}`

RETURNS	DESCRIPTION
`list[str]`	List of violated invariant names

`on_violation(handler)` ¶

Register a violation handler.

Thread-safe: Protected by RLock.

PARAMETER	DESCRIPTION
`handler`	Function to call on violation TYPE: `Callable[[Violation], None]`

`get_violations(since=None, severity=None)` ¶

Get recorded violations.

Thread-safe: Returns a snapshot copy of violations.

PARAMETER	DESCRIPTION
`since`	Only violations after this timestamp TYPE: `float \| None` DEFAULT: `None`
`severity`	Filter by severity level TYPE: `ViolationSeverity \| None` DEFAULT: `None`

RETURNS	DESCRIPTION
`list[Violation]`	List of matching violations

`clear_violations()` ¶

Clear all recorded violations.

Thread-safe: Protected by RLock.

`enable()` ¶

Enable invariant checking.

Thread-safe: Protected by RLock.

`disable()` ¶

Disable invariant checking.

Thread-safe: Protected by RLock.

`stats()` ¶

Get checking statistics.

Thread-safe: Returns a snapshot of statistics.

RETURNS	DESCRIPTION
`dict[str, Any]`	Dictionary with check and violation counts

`record_violation(violation)` ¶

Record a violation directly.

Thread-safe: Protected by RLock. Used by assert_invariant for direct violation recording.

PARAMETER	DESCRIPTION
`violation`	The violation to record. TYPE: `Violation`

`StateTransitionMonitor` ¶

Monitors state transitions and checks invariants.

Useful for verifying state machine correctness.

Thread Safety

All mutable state is protected by RLock for safe concurrent access.

`current_state` `property` ¶

Get current state.

Thread-safe: Protected by RLock.

`init(name='state_monitor', valid_transitions=None)` ¶

Initialize monitor.

PARAMETER	DESCRIPTION
`name`	Monitor name TYPE: `str` DEFAULT: `'state_monitor'`
`valid_transitions`	Map of state to valid next states TYPE: `dict[str, set[str]] \| None` DEFAULT: `None`

`set_valid_transitions(transitions)` ¶

Set valid state transitions.

Thread-safe: Protected by RLock.

PARAMETER	DESCRIPTION
`transitions`	Map of state to valid next states TYPE: `dict[str, set[str]]`

`transition(new_state)` ¶

Record a state transition.

Thread-safe: Protected by RLock.

PARAMETER	DESCRIPTION
`new_state`	The new state TYPE: `str`

RETURNS	DESCRIPTION
`bool`	True if transition is valid

`get_history()` ¶

Get transition history.

Thread-safe: Returns a copy of the history.

RETURNS	DESCRIPTION
`list[tuple[str, str, float]]`	List of (from_state, to_state, timestamp) tuples

`get_invalid_transitions()` ¶

Get invalid transitions that occurred.

Thread-safe: Returns a copy of invalid transitions.

RETURNS	DESCRIPTION
`list[tuple[str, str]]`	List of (from_state, to_state) tuples

`reset()` ¶

Reset monitor state.

Thread-safe: Protected by RLock.

`register_invariant(name, condition, severity=ViolationSeverity.ERROR, message='', registry=None)` ¶

Register an invariant in the default or specified registry.

PARAMETER	DESCRIPTION
`name`	Unique name for invariant TYPE: `str`
`condition`	Function returning True if invariant holds TYPE: `Callable[..., bool]`
`severity`	Severity level if violated TYPE: `ViolationSeverity` DEFAULT: `ERROR`
`message`	Message to log on violation TYPE: `str` DEFAULT: `''`
`registry`	Registry to use (default if None) TYPE: `InvariantRegistry \| None` DEFAULT: `None`

RETURNS	DESCRIPTION
`Invariant`	The registered Invariant

`check_invariant(name, *args, context=None, registry=None, **kwargs)` ¶

Check an invariant in the default or specified registry.

PARAMETER	DESCRIPTION
`name`	Invariant name to check TYPE: `str`
`*args`	Arguments to pass to invariant condition TYPE: `Any` DEFAULT: `()`
`context`	Additional context for violation reporting TYPE: `dict[str, Any] \| None` DEFAULT: `None`
`registry`	Registry to use (default if None) TYPE: `InvariantRegistry \| None` DEFAULT: `None`
`**kwargs`	Keyword arguments to pass to condition TYPE: `Any` DEFAULT: `{}`

RETURNS	DESCRIPTION
`bool`	True if invariant holds

`invariant(name, condition=None, severity=ViolationSeverity.ERROR, message='')` ¶

Decorator to add invariant checking to a function.

The invariant is checked before and after the function executes.

PARAMETER	DESCRIPTION
`name`	Invariant name TYPE: `str`
`condition`	Condition to check (receives function args) TYPE: `Callable[..., bool] \| None` DEFAULT: `None`
`severity`	Severity level if violated TYPE: `ViolationSeverity` DEFAULT: `ERROR`
`message`	Message to log on violation TYPE: `str` DEFAULT: `''`

RETURNS	DESCRIPTION
`Callable[[Callable[P, R]], Callable[P, R]]`	Decorated function

Example

@invariant( "positive_balance", lambda self: self.balance >= 0, message="Balance cannot be negative" ) def withdraw(self, amount): self.balance -= amount

`assert_invariant(condition, message='Assertion failed', severity=ViolationSeverity.ERROR)` ¶

Assert an invariant condition.

Thread-safe: Uses thread-safe record_violation method.

PARAMETER	DESCRIPTION
`condition`	Condition that should be true TYPE: `bool`
`message`	Message if condition is false TYPE: `str` DEFAULT: `'Assertion failed'`
`severity`	Severity of violation TYPE: `ViolationSeverity` DEFAULT: `ERROR`

RAISES	DESCRIPTION
`AssertionError`	If condition is false and severity is FATAL

agenticraft_foundation.verification.invariant_checker¶

ViolationSeverity ¶

Invariant dataclass ¶

check(*args, **kwargs) ¶

Violation dataclass ¶

InvariantRegistry ¶

__init__(name='default') ¶

register(name, condition, severity=ViolationSeverity.ERROR, message='') ¶

unregister(name) ¶

check(name, *args, context=None, **kwargs) ¶

check_all(*args, context=None, **kwargs) ¶

on_violation(handler) ¶

get_violations(since=None, severity=None) ¶

clear_violations() ¶

enable() ¶

disable() ¶

stats() ¶

record_violation(violation) ¶

StateTransitionMonitor ¶

current_state property ¶

__init__(name='state_monitor', valid_transitions=None) ¶

set_valid_transitions(transitions) ¶

transition(new_state) ¶

get_history() ¶

get_invalid_transitions() ¶

reset() ¶

register_invariant(name, condition, severity=ViolationSeverity.ERROR, message='', registry=None) ¶

check_invariant(name, *args, context=None, registry=None, **kwargs) ¶

invariant(name, condition=None, severity=ViolationSeverity.ERROR, message='') ¶

assert_invariant(condition, message='Assertion failed', severity=ViolationSeverity.ERROR) ¶

`ViolationSeverity` ¶

`Invariant` `dataclass` ¶

`check(*args, **kwargs)` ¶

`Violation` `dataclass` ¶

`InvariantRegistry` ¶

`init(name='default')` ¶

`register(name, condition, severity=ViolationSeverity.ERROR, message='')` ¶

`unregister(name)` ¶

`check(name, *args, context=None, **kwargs)` ¶

`check_all(*args, context=None, **kwargs)` ¶

`on_violation(handler)` ¶

`get_violations(since=None, severity=None)` ¶

`clear_violations()` ¶

`enable()` ¶

`disable()` ¶

`stats()` ¶

`record_violation(violation)` ¶

`StateTransitionMonitor` ¶

`current_state` `property` ¶

`init(name='state_monitor', valid_transitions=None)` ¶

`set_valid_transitions(transitions)` ¶

`transition(new_state)` ¶

`get_history()` ¶

`get_invalid_transitions()` ¶

`reset()` ¶

`register_invariant(name, condition, severity=ViolationSeverity.ERROR, message='', registry=None)` ¶

`check_invariant(name, *args, context=None, registry=None, **kwargs)` ¶

`invariant(name, condition=None, severity=ViolationSeverity.ERROR, message='')` ¶

`assert_invariant(condition, message='Assertion failed', severity=ViolationSeverity.ERROR)` ¶