Agent Hypervisor

pip install agent-hypervisor

from hypervisor import Hypervisor, SessionConfig, ConsistencyMode

hv = Hypervisor()

# Create an isolated session with governance
session = await hv.create_session(
    config=SessionConfig(enable_audit=True),
    creator_did="did:mesh:admin",
)

# Agent joins — ring assigned automatically by trust score
ring = await hv.join_session(
    session.sso.session_id,
    "did:mesh:agent-1",
    sigma_raw=0.85,
)
# → RING_2_STANDARD (trusted agent)

# Activate and run a governed saga
await hv.activate_session(session.sso.session_id)
saga = session.saga.create_saga(session.sso.session_id)
step = session.saga.add_step(
    saga.saga_id, "draft_email", "did:mesh:agent-1",
    execute_api="/api/draft", undo_api="/api/undo-draft",
    timeout_seconds=30, max_retries=2,
)
result = await session.saga.execute_step(
    saga.saga_id, step.step_id, executor=draft_email,
)

# Terminate — returns tamper-evident audit hash
hash_root = await hv.terminate_session(session.sso.session_id)

from hypervisor import Hypervisor, SessionConfig, ConsistencyMode, ExecutionRing

# Initialize with optional liability cap and retention policy
hv = Hypervisor(
    max_exposure=1000.0,          # Max total liability per voucher
    retention_policy=None,        # Ephemeral GC rules (default: keep all)
)

# Create a session with resource limits
session = await hv.create_session(
    config=SessionConfig(
        consistency_mode=ConsistencyMode.EVENTUAL,  # or STRONG
        max_participants=10,           # 1–1000
        max_duration_seconds=3600,     # 1–604,800 (7 days max)
        min_eff_score=0.60,            # Minimum trust score to join
        enable_audit=True,             # Hash-chained audit trail
        enable_blockchain_commitment=False,
    ),
    creator_did="did:mesh:admin",
)

# Agent joins — ring assigned by trust score
ring = await hv.join_session(
    session.sso.session_id,
    "did:mesh:agent-1",
    sigma_raw=0.85,   # Raw trust score [0.0–1.0]
)
# Ring assignment thresholds:
#   eff_score > 0.95 + consensus → RING_1_PRIVILEGED
#   eff_score > 0.60             → RING_2_STANDARD
#   otherwise                    → RING_3_SANDBOX (default)

from hypervisor import RingElevationManager

elevation_mgr = RingElevationManager()

# Grant temporary Ring 1 access (max 3600s, default 300s)
elevation = elevation_mgr.elevate(
    agent_did="did:mesh:agent-1",
    session_id=session.sso.session_id,
    target_ring=ExecutionRing.RING_1_PRIVILEGED,
    ttl_seconds=300,              # Auto-expires after 5 minutes
    reason="deploy-approval",
    attestation="signed-by-sre",  # Optional proof
)

# Revoke early if needed
elevation_mgr.revoke(elevation.elevation_id)

from hypervisor import SessionConfig, ConsistencyMode

config = SessionConfig(
    consistency_mode=ConsistencyMode.STRONG,  # Requires consensus
    max_participants=5,
    max_duration_seconds=7200,    # 2-hour session
    min_eff_score=0.70,           # Higher trust threshold
    enable_audit=True,
    enable_blockchain_commitment=True,
)

session = await hv.create_session(config=config, creator_did="did:mesh:admin")
await hv.activate_session(session.sso.session_id)

# Session lifecycle: CREATED → HANDSHAKING → ACTIVE → TERMINATING → ARCHIVED

from hypervisor import SagaDSLParser, SagaOrchestrator, FanOutPolicy

# Option 1: Define saga as a dict (or load from YAML)
definition = {
    "name": "deploy-pipeline",
    "session_id": "ss-a1b2c3d4",
    "steps": [
        {
            "id": "provision",
            "action_id": "provision-vm",
            "agent": "did:mesh:agent-1",
            "execute_api": "/infra/provision",
            "undo_api": "/infra/deprovision",   # Compensation endpoint
            "timeout": 120,                      # Seconds (default: 300)
            "retries": 2,                        # Retry count (default: 0)
        },
        {
            "id": "deploy",
            "action_id": "deploy-app",
            "agent": "did:mesh:agent-2",
            "execute_api": "/app/deploy",
            "undo_api": "/app/undeploy",
            "timeout": 60,
        },
    ],
    "fan_outs": [
        {
            "policy": "all_must_succeed",        # or majority_must_succeed, any_must_succeed
            "branch_step_ids": ["provision", "deploy"],
        },
    ],
}

parser = SagaDSLParser()
errors = parser.validate(definition)   # Returns [] if valid
saga_def = parser.parse(definition)
steps = parser.to_saga_steps(saga_def)

# Option 2: Build programmatically
saga = session.saga.create_saga(session.sso.session_id)
step = session.saga.add_step(
    saga.saga_id, "draft_email", "did:mesh:agent-1",
    execute_api="/api/draft",
    undo_api="/api/undo-draft",
    timeout_seconds=30,
    max_retries=2,
)
result = await session.saga.execute_step(
    saga.saga_id, step.step_id, executor=draft_email,
)
# On failure: automatic reverse-order compensation of committed steps

from hypervisor import KillSwitch

kill_switch = KillSwitch()

# Terminate a misbehaving agent
result = kill_switch.kill(
    agent_did="did:mesh:rogue-agent",
    session_id=session.sso.session_id,
    reason="ring_breach",       # behavioral_drift | rate_limit | ring_breach | manual
)
# result.handoffs — list of in-flight saga steps handed to substitute agents
# result.compensation_triggered — True if active sagas were compensated

from hypervisor import AgentRateLimiter
from hypervisor.rings import ExecutionRing

limiter = AgentRateLimiter()

# Default per-ring limits (rate tokens/sec, burst capacity):
#   Ring 0 (Root):       100.0 rate, 200.0 capacity
#   Ring 1 (Privileged):  50.0 rate, 100.0 capacity
#   Ring 2 (Standard):    20.0 rate,  40.0 capacity
#   Ring 3 (Sandbox):      5.0 rate,  10.0 capacity

# Custom rate limits per ring
from hypervisor.security.rate_limiter import DEFAULT_RING_LIMITS
custom_limits = {
    ExecutionRing.RING_0_ROOT: (200.0, 400.0),
    ExecutionRing.RING_1_PRIVILEGED: (100.0, 200.0),
    ExecutionRing.RING_2_STANDARD: (30.0, 60.0),
    ExecutionRing.RING_3_SANDBOX: (2.0, 5.0),
}
limiter = AgentRateLimiter(ring_limits=custom_limits)

from hypervisor import RingBreachDetector, BreachSeverity

detector = RingBreachDetector()

# Breach events include:
#   severity: NONE | LOW | MEDIUM | HIGH | CRITICAL
#   anomaly_score: float — how far the behavior deviates
#   actual_rate vs expected_rate — call frequency anomaly
#   call_count_window — calls in the detection window

# Breach detection triggers automatic demotion or kill switch

# saga-deploy.yaml
name: deploy-pipeline
session_id: ss-a1b2c3d4
steps:
  - id: provision
    action_id: provision-vm
    agent: "did:mesh:agent-1"
    execute_api: /infra/provision
    undo_api: /infra/deprovision
    timeout: 120
    retries: 2

  - id: deploy
    action_id: deploy-app
    agent: "did:mesh:agent-2"
    execute_api: /app/deploy
    undo_api: /app/undeploy
    timeout: 60
    retries: 1

fan_outs:
  - policy: all_must_succeed
    branch_step_ids:
      - provision
      - deploy

metadata:
  environment: production
  owner: platform-team

import yaml
from hypervisor import SagaDSLParser

with open("saga-deploy.yaml") as f:
    definition = yaml.safe_load(f)

parser = SagaDSLParser()
errors = parser.validate(definition)
if not errors:
    saga_def = parser.parse(definition)

# docker-compose.yml
services:
  redis:
    image: redis:7-alpine
    ports:
      - "6379:6379"

  hypervisor-api:
    build: .
    environment:
      - REDIS_URL=redis://redis:6379/0
      - HYPERVISOR_CONFIG=/app/config/hypervisor.yaml
    ports:
      - "8000:8000"
    volumes:
      - ./config:/app/config

graph TD
    R0["🔴 Ring 0 — Root<br/>Hypervisor config & penalty<br/>Requires SRE Witness"]
    R1["🟠 Ring 1 — Privileged<br/>Non-reversible actions<br/>eff_score > 0.95 + consensus"]
    R2["🟡 Ring 2 — Standard<br/>Reversible actions<br/>eff_score > 0.60"]
    R3["🟢 Ring 3 — Sandbox<br/>Read-only / research<br/>Default for unknown agents"]

    R0 -->|"supervises"| R1
    R1 -->|"supervises"| R2
    R2 -->|"supervises"| R3

stateDiagram-v2
    [*] --> Ring3 : Agent joins session
    Ring3 --> Ring2 : eff_score rises above 0.60
    Ring2 --> Ring1 : eff_score > 0.95 + consensus
    Ring1 --> Ring0 : SRE Witness approval

    Ring0 --> Ring1 : Trust drops / TTL expires
    Ring1 --> Ring2 : Trust drops below 0.95
    Ring2 --> Ring3 : Trust drops below 0.60
    Ring3 --> [*] : Terminated / expelled

    Ring2 --> Ring1 : Sudo elevation (TTL)
    Ring1 --> Ring2 : TTL expires

    note right of Ring3 : Ring breach detection\ntriggers immediate demotion

flowchart LR
    Create["Create Saga"] --> AddSteps["Add Steps"]
    AddSteps --> Execute["Execute Steps"]
    Execute --> Success{"All steps\nsucceed?"}
    Success -- Yes --> Complete["✅ Saga Complete"]
    Success -- No --> Compensate["Compensate\n(reverse order)"]
    Compensate --> CompOk{"Compensation\nsucceeds?"}
    CompOk -- Yes --> Rolled["↩️ Saga Rolled Back"]
    CompOk -- No --> Escalate["⚠️ Escalate\nLiability Penalty"]

flowchart TD
    Sponsor["🛡️ Sponsor Agent<br/>eff_score: 0.92<br/>Bonds reputation"]
    Sponsored["🤖 Sponsored Agent<br/>eff_score: 0.45<br/>Gains Ring 2 access"]
    Action["Agent performs action"]
    Check{"Intent\nviolation?"}
    Safe["✅ No penalty"]
    Penalty["🔻 Both penalized<br/>Sponsor collateral slashed<br/>Sponsored demoted"]

    Sponsor -->|"vouches for"| Sponsored
    Sponsored --> Action
    Action --> Check
    Check -- No --> Safe
    Check -- Yes --> Penalty
    Penalty -->|"collateral slash"| Sponsor
    Penalty -->|"demotion + quarantine"| Sponsored

flowchart TD
    Violation["🚨 Violation Detected"]
    Attr["Fault Attribution<br/>Identify responsible agent"]
    Primary["Primary Agent<br/>Full penalty applied"]
    Sponsor1["Sponsor A<br/>Collateral slashed"]
    Sponsor2["Sponsor B<br/>Collateral slashed"]
    Quarantine["Quarantine Agent<br/>Before termination"]
    Demote["Demote to Ring 3"]
    Ledger["Record in<br/>Liability Ledger"]

    Violation --> Attr
    Attr --> Primary
    Primary --> Sponsor1
    Primary --> Sponsor2
    Primary --> Quarantine
    Quarantine --> Demote
    Sponsor1 --> Ledger
    Sponsor2 --> Ledger
    Primary --> Ledger

pip install agent-hypervisor

# Run all tests
pytest tests/ -v

# Run only integration tests
pytest tests/integration/ -v

# Run benchmarks
python agent-governance-python/benchmarks/bench_hypervisor.py

pip install agent-hypervisor[api]
uvicorn hypervisor.api.server:app
# Open http://localhost:8000/docs for Swagger UI

cd examples/dashboard
pip install -r requirements.txt
streamlit run app.py

graph TB
    subgraph Ecosystem["Agent Governance Ecosystem"]
        OS["🧠 Agent OS<br/>Policy Enforcement Kernel"]
        Mesh["🔗 Agent Mesh<br/>Cryptographic Trust Network"]
        SRE["📊 Agent SRE<br/>Reliability Platform"]
        HV["⚡ Agent Hypervisor<br/>Runtime Governance"]

        OS <-->|"policies"| HV
        Mesh <-->|"trust scores"| HV
        SRE <-->|"SLOs + chaos"| HV
        OS <-->|"identity"| Mesh
    end

    style HV fill:#ff6b6b,stroke:#333,color:#fff