kovrin-safety 0.2.0

Safety middleware for AI agents — constitutional checks, risk routing, cryptographic audit

kovrin-safety

Provable safety infrastructure for AI agents.
Constitutional checks. Cryptographic audit. Reliability engineering. One package.

---

AI agents can browse the web, write code, move money, and delete data. The frameworks that power them — LangGraph, CrewAI, AutoGen — have zero architectural safety guarantees. No formal constraints. No tamper-evident audit. No reliability engineering.

kovrin-safety is the missing layer.

from kovrin_safety import KovrinSafety

safety = KovrinSafety()
result = safety.check_sync("Delete all user data", risk_level="CRITICAL")

result.approved        # False
result.action          # HUMAN_APPROVAL
result.reasoning       # "Harm Floor violation: action matches destructive pattern"

3 lines. No API key. No LLM. Works offline. Works in CI. Works in production.

---

What makes this different

Most "AI safety" tools are glorified string filters or prompt wrappers. kovrin-safety is infrastructure — the kind you'd build for a bank or a hospital.

	kovrin-safety	Guardrails AI	NeMo Guardrails	LangGraph
Constitutional axioms (formal constraints)	5 axioms, SHA-256 integrity	Validators	Colang rules	None
Deterministic risk routing	4x3 matrix, CRITICAL hardcoded	Binary pass/fail	Binary pass/fail	None
Cryptographic audit trail	Merkle hash chain, tamper-evident	Basic logging	None	LangSmith (SaaS)
Persistent audit (survives restart)	SQLite backend	None	None	None
Semantic safety (catches paraphrases)	Local embeddings, zero API	LLM-dependent	LLM-dependent	None
Reliability engineering	Circuit breaker, retry, drift detection	None	None	None
Works without LLM	Yes (rule-based + embeddings)	Partial	No	N/A
Formal verification specs	8 TLA+ modules	None	None	None
Safety metrics & observability	Built-in counters, exportable	None	None	LangSmith (SaaS)

---

The Safety Stack

kovrin-safety gives you 6 layers of protection. Use all of them, or just the ones you need.

Layer 0: Constitutional Core

5 immutable axioms validated before every action. Protected by SHA-256 integrity hash — they cannot be modified at runtime.

Axiom	Guarantee
Human Agency	No action removes the ability for humans to override, pause, or terminate
Harm Floor	Expected harm never exceeds the defined threshold
Transparency	All decisions are traceable to the original intent
Reversibility	Irreversible actions require explicit human approval
Scope Limit	Never exceed the authorized operational boundary

Three checker modes — choose your trade-off:

from kovrin_safety.constitutional import create_checker

# Rule-based (fast, zero dependencies, catches obvious attacks)
checker = create_checker(mode="rule")

# Embedding-based (catches paraphrases regex misses, local, no API)
checker = create_checker(mode="embedding")

# Claude-powered (full semantic analysis, requires API key)
checker = create_checker(api_key="sk-ant-...", mode="claude")

The embedding checker uses all-MiniLM-L6-v2 locally. It catches what regex cannot:

"Delete all user data"                              -> Blocked (regex catches this)
"Remove every customer record from the system"      -> Blocked (only embeddings catch this)
"Erase the entire client database permanently"      -> Blocked (only embeddings catch this)
"Generate a summary report of monthly sales"        -> Allowed

Layer 1: Risk Router

Deterministic 4x3 routing matrix. No randomness. No LLM judgment. Pure logic.

              FREE          GUARDED        NONE
  LOW       Auto Execute   Auto Execute   Sandbox Review
  MEDIUM    Auto Execute   Sandbox Review Human Approval
  HIGH      Sandbox Review Human Approval Human Approval
  CRITICAL  Human Approval Human Approval Human Approval

Safety invariant: CRITICAL always routes to HUMAN_APPROVAL. This is hardcoded. No profile, no override, no configuration can change this. Ever.

4 autonomy profiles control the posture:

KovrinSafety(profile="DEFAULT")     # Balanced
KovrinSafety(profile="CAUTIOUS")    # Stricter — more human approvals
KovrinSafety(profile="AGGRESSIVE")  # Relaxed — CRITICAL still blocked
KovrinSafety(profile="LOCKED")      # Everything requires human approval

Layer 2: Critic Pipeline

Three critics evaluate every action. All must pass — one failure blocks the action.

Critic	What it checks	LLM required?
SafetyCritic	Constitutional axiom compliance	No
FeasibilityCritic	"Is this action actually doable?"	Optional
PolicyCritic	Organization-specific constraints	Optional

safety = KovrinSafety(
    constraints=[
        "Do not delete customer data",
        "Never share confidential information externally",
        "All financial transactions require manager approval",
    ]
)

Layer 3: Cryptographic Audit Trail

Every safety check is recorded in an append-only Merkle hash chain. SHA-256 linked. Tamper-evident. Exportable.

# Persistent audit — survives process restart
safety = KovrinSafety(audit_storage_path="./safety_audit.db")

await safety.check("Action 1", risk_level="LOW")
await safety.check("Action 2", risk_level="HIGH")

# Verify chain integrity (detects any tampering)
assert safety.verify_integrity() is True

# Export for compliance (EU AI Act Articles 9, 12, 14, 15)
report = safety.export_compliance_report()

Without audit_storage_path, the chain lives in memory (same as before). With it, every event is persisted to SQLite in WAL mode — append-only, no delete, no modify.

Layer 4: Watchdog Monitor

Real-time behavioral monitoring with graduated containment: WARN -> PAUSE -> KILL.

Rule	Trigger	Action
NoExecutionAfterRejection	Previously rejected task runs again	KILL
ExcessiveFailureRate	Success rate drops below threshold	PAUSE
UnexpectedEventSequence	Out-of-order events	WARN
ExcessiveToolCallRate	Rate limit violation	PAUSE
ToolEscalationDetection	Privilege escalation attempt	WARN
ToolCallAfterBlock	Repeated bypass attempts	PAUSE

safety = KovrinSafety(enable_watchdog=True)

Layer 5: Reliability Engineering

Production AI systems fail in ways traditional software doesn't. LLM APIs go down. Agent outputs drift. Responses become inconsistent. kovrin-safety includes purpose-built reliability primitives.

Circuit Breaker

from kovrin_safety.reliability import CircuitBreaker

breaker = CircuitBreaker(failure_threshold=5, recovery_timeout=30.0)

# Wraps any async call — opens after 5 failures, auto-recovers
result = await breaker.call(lambda: llm_provider.generate(prompt))

breaker.state  # CLOSED -> OPEN -> HALF_OPEN -> CLOSED

HALF_OPEN state uses single-probe serialization — only one test call is allowed through while others are rejected, preventing thundering herd on recovery.

Retry with Backoff

from kovrin_safety.reliability import RetryPolicy

policy = RetryPolicy(
    max_retries=3,
    base_delay=1.0,
    max_delay=60.0,
    retry_on=(ConnectionError, TimeoutError),
)

result = await policy.execute(lambda: api_call())

Exponential backoff with full jitter (AWS best practice) to prevent retry storms.

Fallback Chains

from kovrin_safety.reliability import FallbackChain

chain = FallbackChain([
    lambda: claude_provider.generate(prompt),
    lambda: openai_provider.generate(prompt),
    lambda: local_model.generate(prompt),
])

result = await chain.execute()  # Tries each until one succeeds

Timeout Budgets

from kovrin_safety.reliability import TimeoutBudget

budget = TimeoutBudget(total_seconds=30.0)

step1 = await budget.run(lambda: fetch_context())      # Uses part of budget
step2 = await budget.run(lambda: generate_response())   # Uses remaining
# If budget runs out -> TimeoutError

Drift Detection

Embedding-based semantic drift monitoring. Detects when agent outputs diverge from the original intent.

from kovrin_safety.reliability import DriftDetector

detector = DriftDetector(drift_threshold=0.5)
detector.set_reference("Analyze customer support tickets and summarize trends")

signal = detector.check("Found 50 open tickets about billing issues")
signal.drifted  # False — on topic

signal = detector.check("Bitcoin reached new highs today")
signal.drifted  # True — topic drift detected
signal.score    # 0.82 — high drift score

All-in-One Integration

from kovrin_safety import KovrinSafety, ReliabilityConfig

safety = KovrinSafety(
    profile="cautious",
    audit_storage_path="./audit.db",
    reliability=ReliabilityConfig(
        fault_tolerance=True,    # Circuit breaker + retry ready
        drift_detection=True,    # Embedding-based output monitoring
    ),
)

# Access reliability components
safety.circuit_breaker.state    # CircuitState.CLOSED
safety.drift_detector           # DriftDetector instance

---

Framework Integrations

LangGraph / LangChain

from kovrin_safety import KovrinSafety
from kovrin_safety.exceptions import SafetyBlockedError
from kovrin_safety.integrations.langchain import KovrinSafetyMiddleware

safety = KovrinSafety(profile="CAUTIOUS")
middleware = KovrinSafetyMiddleware(safety, block_on_rejection=True)

# Pre-model safety gate
state = {"messages": [{"content": "Delete all databases"}]}
try:
    state = await middleware.before_model(state)
except SafetyBlockedError as e:
    print(f"Blocked: {e.failed_critics}")

# Tool call safety gate
result = await middleware.wrap_tool_call(
    tool_name="file_write",
    tool_input={"path": "/etc/passwd"},
)

CrewAI

from kovrin_safety import KovrinSafety
from kovrin_safety.integrations.crewai import kovrin_guardrail

safety = KovrinSafety()

task = Task(
    description="Send marketing emails",
    agent=marketer,
    guardrails=[kovrin_guardrail(safety, risk_level="MEDIUM")],
)

---

Install

pip install kovrin-safety

Optional extras:

pip install kovrin-safety[embeddings]   # Local semantic safety (sentence-transformers)
pip install kovrin-safety[anthropic]    # Claude-powered semantic checks
pip install kovrin-safety[langchain]    # LangGraph integration
pip install kovrin-safety[crewai]       # CrewAI integration
pip install kovrin-safety[all]          # Everything

Core requirements: Python 3.12+, pydantic >= 2.0. That's it.

---

API Reference

KovrinSafety

KovrinSafety(
    profile="DEFAULT",                    # DEFAULT | CAUTIOUS | AGGRESSIVE | LOCKED
    anthropic_api_key=None,               # Enables Claude-powered critics
    constraints=None,                     # Organization policy constraints
    enable_watchdog=False,                # Real-time behavioral monitoring
    audit_storage_path=None,              # SQLite path for persistent audit trail
    reliability=None,                     # ReliabilityConfig for fault tolerance + drift
)

Method	Returns	Description
check(action, risk_level, ...)	SafetyResult	Full async safety pipeline
check_sync(action, risk_level, ...)	SafetyResult	Synchronous wrapper
verify_integrity()	bool	Verify Merkle chain integrity
export_compliance_report()	dict	EU AI Act compliance report
metrics()	dict	Safety metrics (checks, approvals, rejections, risk distribution)

Property	Type	Description
audit_log	ImmutableTraceLog	Access the audit trail
watchdog	WatchdogMonitor | None	Access the watchdog
circuit_breaker	CircuitBreaker | None	Access the circuit breaker
drift_detector	DriftDetector | None	Access the drift detector

SafetyResult

SafetyResult(
    approved: bool,                           # Can the action proceed?
    action: RoutingAction,                    # AUTO_EXECUTE | SANDBOX_REVIEW | HUMAN_APPROVAL
    risk_level: RiskLevel,                    # LOW | MEDIUM | HIGH | CRITICAL
    speculation_tier: SpeculationTier,        # FREE | GUARDED | NONE
    proof_obligations: list[ProofObligation], # Evidence from all critics
    trace_id: str,                            # Unique identifier for this check
    reasoning: str,                           # Human-readable explanation
    timestamp: datetime,                      # When the check ran
)

---

Development

git clone https://github.com/nkovalcin/kovrin-safety.git
cd kovrin-safety
python3.12 -m venv .venv && source .venv/bin/activate
pip install -e ".[dev]"

pytest tests/ -v          # 522 tests
ruff check src/ tests/    # Lint
mypy src/kovrin_safety/   # Type check (strict)

Architecture

src/kovrin_safety/
├── middleware.py              # KovrinSafety — main orchestrator
├── constitutional.py         # Layer 0: 5 axioms, SHA-256 integrity
├── embeddings.py             # Semantic checker (sentence-transformers)
├── critics.py                # SafetyCritic, FeasibilityCritic, PolicyCritic
├── risk_router.py            # Deterministic 4x3 routing matrix
├── audit.py                  # Merkle hash chain (in-memory + SQLite)
├── storage.py                # SQLite persistent backend
├── watchdog.py               # 6 temporal rules, graduated containment
├── models.py                 # Pydantic models + enums
├── exceptions.py             # Exception hierarchy
├── metrics.py                # Safety metrics counters
├── thresholds.py             # Configurable drift thresholds
├── reliability/
│   ├── config.py             # ReliabilityConfig
│   ├── fault_tolerance.py    # CircuitBreaker, RetryPolicy, FallbackChain, TimeoutBudget
│   └── drift_detector.py     # Embedding-based output drift detection
└── integrations/
    ├── langchain.py          # LangGraph/LangChain middleware
    └── crewai.py             # CrewAI guardrails

---

Part of the Kovrin Ecosystem

kovrin-safety is the standalone safety middleware extracted from Kovrin — a safety-first AI agent orchestration framework with 1,000+ tests and 8 TLA+ formal verification modules.

Kovrin provides the full orchestration engine (intent parsing, graph execution, MCTS exploration, speculative execution). kovrin-safety provides just the safety layer — plug it into any existing pipeline.

Learn more at kovrin.dev | Documentation | Dashboard

---

License

MIT — see LICENSE.

Built by Norbert Kovalcin at DIGITAL SPECIALISTS s.r.o.