Metadata-Version: 2.4
Name: kovrin-safety
Version: 0.2.0
Summary: Safety middleware for AI agents — constitutional checks, risk routing, cryptographic audit
Project-URL: Homepage, https://kovrin.dev
Project-URL: Repository, https://github.com/nkovalcin/kovrin-safety
Project-URL: Documentation, https://docs.kovrin.dev
Project-URL: Issues, https://github.com/nkovalcin/kovrin-safety/issues
Author-email: Norbert Kovalčín <norbert@nkovalcin.com>
License-Expression: MIT
License-File: LICENSE
Keywords: agents,ai-safety,audit,constitutional-ai,middleware
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Topic :: Scientific/Engineering :: Artificial Intelligence
Classifier: Topic :: Security
Requires-Python: >=3.12
Requires-Dist: pydantic>=2.0
Provides-Extra: all
Requires-Dist: anthropic>=0.40.0; extra == 'all'
Requires-Dist: crewai>=0.80; extra == 'all'
Requires-Dist: langchain-core>=0.3; extra == 'all'
Requires-Dist: sentence-transformers>=3.0; extra == 'all'
Provides-Extra: anthropic
Requires-Dist: anthropic>=0.40.0; extra == 'anthropic'
Provides-Extra: crewai
Requires-Dist: crewai>=0.80; extra == 'crewai'
Provides-Extra: dev
Requires-Dist: mypy>=1.13; extra == 'dev'
Requires-Dist: pytest-asyncio>=0.24; extra == 'dev'
Requires-Dist: pytest>=8.0; extra == 'dev'
Requires-Dist: ruff>=0.8; extra == 'dev'
Provides-Extra: embeddings
Requires-Dist: sentence-transformers>=3.0; extra == 'embeddings'
Provides-Extra: langchain
Requires-Dist: langchain-core>=0.3; extra == 'langchain'
Description-Content-Type: text/markdown

<p align="center">
  <h1 align="center">kovrin-safety</h1>
  <p align="center">
    <strong>Provable safety infrastructure for AI agents.</strong>
    <br />
    Constitutional checks. Cryptographic audit. Reliability engineering. One package.
  </p>
</p>

<p align="center">
  <a href="LICENSE"><img src="https://img.shields.io/badge/License-MIT-green.svg" alt="MIT License" /></a>
  <a href="https://www.python.org/downloads/"><img src="https://img.shields.io/badge/python-3.12+-blue.svg" alt="Python 3.12+" /></a>
  <img src="https://img.shields.io/badge/tests-522%20passed-brightgreen.svg" alt="522 tests" />
  <img src="https://img.shields.io/badge/dependencies-1%20(pydantic)-orange.svg" alt="1 dependency" />
  <img src="https://img.shields.io/badge/LLM%20required-no-success.svg" alt="No LLM required" />
</p>

---

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.

```python
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:

```python
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:

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

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

```python
# 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 |

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

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

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

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

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

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

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

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

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

```bash
pip install kovrin-safety
```

Optional extras:

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

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

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

```bash
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](https://kovrin.dev) — 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](https://kovrin.dev) | [Documentation](https://docs.kovrin.dev) | [Dashboard](https://app.kovrin.dev)

---

## License

MIT — see [LICENSE](LICENSE).

Built by [Norbert Kovalcin](https://nkovalcin.com) at DIGITAL SPECIALISTS s.r.o.
