Moonshot AI + kvcache-ai Open Source AgentENV — A Deep Dive into the Firecracker-Based Agentic RL Environment Scaling Platform

# Traditional RL: environment is an in-memory function
obs, reward, done, info = env.step(action)

# AgentENV Jailer configuration pseudocode
# Actual implementation is in Rust; Python is used for illustration

def configure_jailer(microvm_id: str, resources: ResourceSpec):
    """Configure Firecracker Jailer isolation parameters"""
    jailer_cfg = {
        # 1. Create independent namespaces
        "namespaces": {
            "mount": True,      # Independent mount tree
            "pid": True,        # Independent PID space
            "net": True,        # Independent network stack
            "ipc": True,        # Independent IPC resources
            "uts": True,        # Independent hostname
        },
        # 2. cgroups resource limits
        "cgroups": {
            "cpu_quota": resources.cpu_quota_us,     # Microsecond CPU quota
            "memory_max": resources.memory_mb,        # Maximum memory (including balloon)
            "memory_swap_max": 0,                      # Disable swap
            "io_weight": resources.io_weight,          # IO priority
            "pids_max": resources.max_pids,            # Maximum process count
        },
        # 3. seccomp filter (syscall whitelist)
        "seccomp": {
            "default_action": "KILL",  # Syscalls outside whitelist kill the process
            "allowed_syscalls": [
                # Only the minimal syscall set needed for microVM operation
                "read", "write", "openat", "close",
                "mmap", "munmap", "mprotect",
                "futex", "clock_gettime", "nanosleep",
                "ioctl",   # For KVM interface
                # ... approximately 50 syscalls
            ],
        },
        # 4. chroot to dedicated directory
        "chroot_dir": f"/var/lib/aenv/jailer/{microvm_id}/",
        # 5. Run as non-root user
        "uid": 1000 + hash(microvm_id) % 60000,
        "gid": 1000 + hash(microvm_id) % 60000,
    }
    return jailer_cfg

# ublk + overlaybd I/O flow pseudocode
# Actual implementation uses Rust ublk bindings

def handle_ublk_io(request: UblkRequest):
    """Handle block device I/O request from Guest"""
    sector = request.sector          # Request sector number
    nr_sectors = request.nr_sectors  # Number of sectors requested
    is_write = request.is_write      # Read/write

    # 1. Map sector to overlaybd layer
    layer_info = overlaybd.lookup(sector, nr_sectors)

    if is_write:
        # Write: write to COW upper layer
        # On first write to a sector, copy base layer data first (COW)
        if not cow_layer.has_sector(sector):
            base_data = base_layer.read(sector, nr_sectors)
            cow_layer.write(sector, nr_sectors, base_data)
        # Write new data to COW upper layer
        cow_layer.write(sector, nr_sectors, request.data)
        # Mark sector as dirty
        dirty_bitmap.mark(sector, nr_sectors)
    else:
        # Read: check COW upper layer first, fall back to base layer
        if cow_layer.has_sector(sector):
            data = cow_layer.read(sector, nr_sectors)
        else:
            data = base_layer.read(sector, nr_sectors)

    # 2. Use io_uring for async I/O
    # 3. Use Direct I/O to avoid double caching on host
    return data

// AgentENV ublk queue processing Rust pseudocode
// Async I/O handling based on io_uring

use io_uring::{IoUring, opcode, types};

struct UblkQueue {
    ring: IoUring,
    sqes: Vec<opcode::ReadWrite>,
}

impl UblkQueue {
    /// Process a batch of I/O requests
    async fn process_io_requests(&mut self, requests: Vec<UblkRequest>) {
        for (i, req) in requests.iter().enumerate() {
            let (buf, offset) = self.get_buffer_and_offset(req);
            let sqe = if req.is_write {
                opcode::Write::new(
                    types::Fixed::new(self.fd),
                    buf,
                    offset,
                ).build()
            } else {
                opcode::Read::new(
                    types::Fixed::new(self.fd),
                    buf,
                    offset,
                ).build()
            };
            // Push to io_uring submission queue
            unsafe { self.ring.submission().push(&sqe).unwrap(); }
        }
        // Wait for completion
        self.ring.submit_and_wait(requests.len()).unwrap();
        // Process completion queue
        for cqe in self.ring.completion() {
            let result = cqe.result();
            // Handle completion event
            self.complete_io(cqe.user_data(), result);
        }
    }
}

# Incremental snapshot Python pseudocode
# Actual AgentENV uses Rust calling Firecracker API

class IncrementalSnapshotManager:
    """Incremental snapshot manager"""

    def __init__(self, microvm_id: str):
        self.microvm_id = microvm_id
        self.base_snapshot = None
        self.delta_snapshots = []
        self.last_dirty_bitmap = None

    async def create_snapshot(self, vm: FirecrackerMicroVM):
        """Create an incremental snapshot"""
        # 1. Pause VM (freeze vCPUs)
        await vm.pause()

        # 2. Get dirty page bitmap since last snapshot
        dirty_bitmap = await vm.get_dirty_log()

        if self.base_snapshot is None:
            # First snapshot: record full state
            mem_snapshot = await vm.dump_memory_full()
            disk_snapshot = await vm.dump_disk_full()
            self.base_snapshot = Snapshot(
                memory=mem_snapshot,
                disk=disk_snapshot,
                bitmap=dirty_bitmap,
                timestamp=time.now(),
            )
        else:
            # Incremental snapshot: record only dirty pages
            new_dirty_pages = self._compute_delta(
                dirty_bitmap, self.last_dirty_bitmap
            )
            mem_delta = await vm.dump_memory_regions(new_dirty_pages)
            disk_delta = await vm.dump_disk_changes(new_dirty_pages)
            delta = DeltaSnapshot(
                base_id=self.base_snapshot.id,
                dirty_pages=new_dirty_pages,
                memory_delta=mem_delta,
                disk_delta=disk_delta,
                timestamp=time.now(),
            )
            self.delta_snapshots.append(delta)

        self.last_dirty_bitmap = dirty_bitmap

        # 3. Resume VM
        await vm.resume()

        return self.base_snapshot or self.delta_snapshots[-1]

    async def restore_from_snapshot(self, snapshot_id: str):
        """Restore environment from snapshot"""
        # 1. Load full state from base snapshot
        vm_state = self.base_snapshot.memory.copy()
        disk_state = self.base_snapshot.disk.copy()

        # 2. Apply deltas in order
        for delta in self.delta_snapshots:
            if delta.base_id == self.base_snapshot.id:
                for page_addr, page_data in delta.memory_delta.items():
                    vm_state[page_addr] = page_data
                for block_addr, block_data in delta.disk_delta.items():
                    disk_state[block_addr] = block_data

        # 3. Load into new Firecracker instance
        new_vm = await create_firecracker_vm()
        await new_vm.load_memory(vm_state)
        await new_vm.load_disk(disk_state)
        await new_vm.resume()

        return new_vm

    def _compute_delta(self, current, previous):
        """Compute difference between two bitmaps"""
        if previous is None:
            return current
        delta = Bitmap()
        for i in range(len(current)):
            if current[i] and not previous[i]:
                delta.set(i)
        return delta

#!/usr/bin/env python3
"""AgentENV Firecracker API operations — complete microVM lifecycle management"""

import asyncio
import json
import aiohttp
from dataclasses import dataclass
from typing import Optional

@dataclass
class MicroVMConfig:
    """microVM configuration"""
    kernel_path: str = "/opt/aenv/kernel/vmlinux-6.8"
    rootfs_path: str = "/opt/aenv/images/ubuntu-22.04.ext4"
    vcpu_count: int = 2
    mem_size_mib: int = 4096
    jailer_cfg: Optional[dict] = None

class FirecrackerAPI:
    """Firecracker control plane API wrapper"""

    def __init__(self, fc_socket_path: str):
        self.socket_path = fc_socket_path
        self.session = aiohttp.ClientSession(
            connector=aiohttp.UnixConnector(path=fc_socket_path)
        )

    async def create_vm(self, config: MicroVMConfig) -> dict:
        """Create and start a microVM"""
        # 1. Configure kernel and root filesystem
        await self._put("/boot-source", {
            "kernel_image_path": config.kernel_path,
            "boot_args": "console=ttyS0 reboot=k panic=1 pci=off"
        })

        # 2. Configure root filesystem drive
        await self._put("/drives/rootfs", {
            "drive_id": "rootfs",
            "path_on_host": config.rootfs_path,
            "is_root_device": True,
            "is_read_only": False,
        })

        # 3. Configure vCPU and memory
        await self._put("/machine-config", {
            "vcpu_count": config.vcpu_count,
            "mem_size_mib": config.mem_size_mib,
            "smt": False,
            "track_dirty_pages": True,  # Enable dirty page tracking (needed for snapshots)
        })

        # 4. Configure network interface
        await self._put("/network-interfaces/eth0", {
            "iface_id": "eth0",
            "host_dev_name": "tap0",
            "guest_mac": "02:fc:00:00:00:01",
        })

        # 5. Start microVM
        await self._put("/actions", {
            "action_type": "InstanceStart"
        })

        return {"status": "running", "vm_id": "vm-1"}

    async def create_snapshot(self, snapshot_id: str, mem_path: str,
                              snapshot_path: str) -> dict:
        """Create memory and disk snapshot"""
        # Pause VM first
        await self._put("/actions", {"action_type": "Pause"})

        # Create snapshot
        await self._put("/snapshot/create", {
            "snapshot_type": "Full",
            "snapshot_path": snapshot_path,
            "mem_file_path": mem_path,
            "version": "1.0",
        })

        # Resume VM
        await self._put("/actions", {"action_type": "Resume"})

        return {"snapshot_id": snapshot_id, "status": "created"}

    async def restore_from_snapshot(self, mem_path: str,
                                    snapshot_path: str) -> dict:
        """Restore VM from snapshot"""
        await self._put("/snapshot/load", {
            "snapshot_path": snapshot_path,
            "mem_backend": {
                "backend_type": "File",
                "backend_path": mem_path,
            },
            "resume_vm": True,
        })

        return {"status": "restored"}

    async def _put(self, path: str, body: dict) -> dict:
        """Send PUT request to Firecracker API"""
        url = f"http://localhost{path}"
        async with self.session.put(url, json=body) as resp:
            if resp.status >= 400:
                text = await resp.text()
                raise RuntimeError(f"API error {resp.status}: {text}")
            if resp.status == 204:
                return {}
            return await resp.json()

    async def close(self):
        await self.session.close()

# Usage example
async def main():
    # Create and start a microVM
    fc_api = FirecrackerAPI("/tmp/firecracker.socket")
    vm = await fc_api.create_vm(MicroVMConfig(
        vcpu_count=4,
        mem_size_mib=8192,
    ))
    print(f"VM created successfully: {vm}")

    # Run some tasks...
    await asyncio.sleep(10)

    # Create snapshot
    snap = await fc_api.create_snapshot(
        snapshot_id="snap-001",
        mem_path="/snapshots/snap-001/mem",
        snapshot_path="/snapshots/snap-001/vmstate",
    )
    print(f"Snapshot created successfully: {snap}")

    # Restore from snapshot to a new VM (Fork)
    restored = await fc_api.restore_from_snapshot(
        mem_path="/snapshots/snap-001/mem",
        snapshot_path="/snapshots/snap-001/vmstate",
    )
    print(f"VM restored successfully: {restored}")

    await fc_api.close()

if __name__ == "__main__":
    asyncio.run(main())

#!/usr/bin/env python3
"""Distributed GRPO Agentic RL training example using AgentENV"""

import asyncio
import json
import time
from dataclasses import dataclass, field
from typing import List, Dict, Optional, Any
import numpy as np
import aiohttp
import torch

# ============================================================
# AgentENV Client (E2B Compatible API)
# ============================================================

class AgentENVClient:
    """AgentENV sandbox client using E2B compatible API"""

    def __init__(self, base_url: str = "http://localhost:8000"):
        self.base_url = base_url
        self.session = aiohttp.ClientSession()

    async def create_sandbox(
        self,
        template: str = "ubuntu:22.04",
        vcpu: int = 2,
        memory_mb: int = 4096,
        timeout_s: int = 300,
    ) -> str:
        """Create a new sandbox environment"""
        async with self.session.post(
            f"{self.base_url}/sandboxes",
            json={
                "templateID": template,
                "vcpu": vcpu,
                "memoryMB": memory_mb,
                "timeout": timeout_s,
                "autoPause": True,
            }
        ) as resp:
            data = await resp.json()
            return data["sandboxID"]

    async def exec_command(
        self, sandbox_id: str, command: str
    ) -> Dict[str, Any]:
        """Execute a command in the sandbox"""
        async with self.session.post(
            f"{self.base_url}/sandboxes/{sandbox_id}/execute",
            json={"command": command}
        ) as resp:
            return await resp.json()

    async def write_file(
        self, sandbox_id: str, path: str, content: str
    ) -> None:
        """Write a file to the sandbox"""
        async with self.session.post(
            f"{self.base_url}/sandboxes/{sandbox_id}/files/write",
            json={"path": path, "content": content}
        ) as resp:
            resp.raise_for_status()

    async def read_file(
        self, sandbox_id: str, path: str
    ) -> str:
        """Read a file from the sandbox"""
        async with self.session.get(
            f"{self.base_url}/sandboxes/{sandbox_id}/files/read",
            params={"path": path}
        ) as resp:
            data = await resp.json()
            return data["content"]

    async def pause_sandbox(self, sandbox_id: str) -> None:
        """Pause sandbox (release CPU/memory)"""
        async with self.session.post(
            f"{self.base_url}/sandboxes/{sandbox_id}/pause"
        ) as resp:
            resp.raise_for_status()

    async def resume_sandbox(self, sandbox_id: str) -> None:
        """Resume sandbox"""
        async with self.session.post(
            f"{self.base_url}/sandboxes/{sandbox_id}/resume"
        ) as resp:
            resp.raise_for_status()

    async def fork_sandbox(
        self, sandbox_id: str, count: int = 4
    ) -> List[str]:
        """Fork N child sandboxes from an existing sandbox"""
        async with self.session.post(
            f"{self.base_url}/sandboxes/{sandbox_id}/fork",
            json={"count": count}
        ) as resp:
            data = await resp.json()
            return data["childSandboxIDs"]

    async def create_template(
        self, sandbox_id: str, name: str
    ) -> str:
        """Create a reusable template from a sandbox state"""
        async with self.session.post(
            f"{self.base_url}/templates",
            json={"sandboxID": sandbox_id, "name": name}
        ) as resp:
            data = await resp.json()
            return data["templateID"]

    async def delete_sandbox(self, sandbox_id: str) -> None:
        """Delete a sandbox"""
        async with self.session.delete(
            f"{self.base_url}/sandboxes/{sandbox_id}"
        ) as resp:
            resp.raise_for_status()

    async def close(self):
        await self.session.close()


# ============================================================
# Task Definition: Coding Task Environment
# ============================================================

@dataclass
class CodingTask:
    """Coding task definition"""
    repo_url: str
    branch: str = "main"
    description: str = ""
    test_command: str = "pytest tests/"
    setup_commands: List[str] = field(default_factory=list)

# Example tasks: fix bugs in Python projects
BENCHMARK_TASKS = [
    CodingTask(
        repo_url="https://github.com/demo/fastapi-bug-fix.git",
        branch="main",
        description="Fix SQL injection vulnerability in FastAPI routes",
        test_command="pytest tests/test_security.py -v",
        setup_commands=[
            "pip install -r requirements.txt",
            "pip install pytest",
        ],
    ),
    CodingTask(
        repo_url="https://github.com/demo/react-form-bug.git",
        branch="main",
        description="Fix state update bug in React form validation component",
        test_command="npm test",
        setup_commands=[
            "npm install",
        ],
    ),
]


# ============================================================
# GRPO Trainer
# ============================================================

@dataclass
class GRPOConfig:
    """GRPO training configuration"""
    group_size: int = 8          # Group size for sampling (GRPO's G)
    num_iterations: int = 100    # Number of training iterations
    learning_rate: float = 1e-5
    kl_coef: float = 0.01
    clip_epsilon: float = 0.2
    max_steps_per_task: int = 20
    fork_count: int = 8          # Number of child sandboxes per fork

class GRPOTrainer:
    """GRPO algorithm trainer using AgentENV as environment backend"""

    def __init__(
        self,
        policy_model: Any,
        tokenizer: Any,
        config: GRPOConfig,
        env_client: AgentENVClient,
    ):
        self.policy = policy_model
        self.tokenizer = tokenizer
        self.config = config
        self.env = env_client
        self.optimizer = torch.optim.AdamW(
            self.policy.parameters(),
            lr=config.learning_rate,
        )

    async def setup_environment(self, task: CodingTask) -> str:
        """Build training environment and return sandbox ID"""
        # 1. Create base sandbox
        sandbox_id = await self.env.create_sandbox(
            template="coding-agent:latest",
            vcpu=4,
            memory_mb=8192,
            timeout_s=3600,
        )

        # 2. Clone code repository
        await self.env.exec_command(
            sandbox_id,
            f"git clone --branch {task.branch} {task.repo_url} /workspace"
        )

        # 3. Execute setup commands
        for cmd in task.setup_commands:
            result = await self.env.exec_command(sandbox_id, cmd)
            if result["exitCode"] != 0:
                raise RuntimeError(
                    f"Setup failed: {result['stderr']}"
                )

        return sandbox_id

    async def collect_trajectories(
        self, base_sandbox_id: str, task: CodingTask
    ) -> List[Dict]:
        """Collect multiple trajectories in parallel using Fork"""
        # 1. Fork child sandboxes from base environment
        child_ids = await self.env.fork_sandbox(
            base_sandbox_id,
            count=self.config.fork_count,
        )
        print(f"Forked {len(child_ids)} child sandboxes from {base_sandbox_id}")

        trajectories = []
        tasks = []

        for i, child_id in enumerate(child_ids):
            tasks.append(
                self._rollout_single(child_id, task, i)
            )

        # Execute all rollouts in parallel
        results = await asyncio.gather(*tasks)
        trajectories.extend(results)

        # Clean up child sandboxes
        for child_id in child_ids:
            await self.env.delete_sandbox(child_id)

        return trajectories

    async def _rollout_single(
        self, sandbox_id: str, task: CodingTask, rank: int
    ) -> Dict:
        """Single trajectory sampling"""
        trajectory = {
            "sandbox_id": sandbox_id,
            "rank": rank,
            "steps": [],
            "final_reward": 0.0,
            "success": False,
        }

        current_state = await self._get_env_state(sandbox_id)

        for step in range(self.config.max_steps_per_task):
            # Policy model generates action
            action = await self._policy_generate(current_state)

            # Execute action in sandbox
            result = await self.env.exec_command(sandbox_id, action)

            # Get new state and reward
            next_state = await self._get_env_state(sandbox_id)
            reward = self._compute_reward(result, task)

            trajectory["steps"].append({
                "state": current_state,
                "action": action,
                "result": result,
                "reward": reward,
                "next_state": next_state,
            })

            current_state = next_state

            # Check if task is completed
            if result["exitCode"] == 0 and "ALL TESTS PASSED" in result["stdout"]:
                trajectory["success"] = True
                trajectory["final_reward"] = 10.0
                break

        return trajectory

    async def _get_env_state(self, sandbox_id: str) -> str:
        """Get environment state summary"""
        result = await self.env.exec_command(
            sandbox_id,
            "echo '=== FILES ===' && find . -name '*.py' -newer /tmp/start | head -20 && "
            "echo '=== GIT DIFF ===' && git diff --stat 2>/dev/null || true"
        )
        return result["stdout"]

    async def _policy_generate(self, state: str) -> str:
        """Policy model generates action (code modification) from state"""
        prompt = (
            f"Current codebase state:\n{state}\n\n"
            f"Generate the next shell command to fix the bug. "
            f"Output only the command, no explanation."
        )
        inputs = self.tokenizer(prompt, return_tensors="pt")
        with torch.no_grad():
            outputs = self.policy.generate(
                **inputs,
                max_new_tokens=200,
                temperature=1.0,
                top_p=0.9,
            )
        action = self.tokenizer.decode(outputs[0], skip_special_tokens=True)
        action = action.split("\n")[-1].strip()
        return action

    def _compute_reward(self, result: Dict, task: CodingTask) -> float:
        """Compute reward for action result"""
        reward = 0.0
        if result["exitCode"] == 0:
            reward += 1.0  # Command executed successfully
        if "ALL TESTS PASSED" in result.get("stdout", ""):
            reward += 5.0  # Tests passed
        if "ERROR" in result.get("stderr", ""):
            reward -= 0.5  # Error occurred
        return reward

    async def train_step(self, trajectories: List[Dict]) -> Dict:
        """GRPO training step"""
        # 1. Collect all rewards and states
        all_rewards = []
        all_states = []
        all_actions = []

        for traj in trajectories:
            for step in traj["steps"]:
                all_rewards.append(step["reward"])
                all_states.append(step["state"])
                all_actions.append(step["action"])

        if not all_rewards:
            return {"loss": 0.0, "mean_reward": 0.0}

        # 2. GRPO: group-normalized rewards
        rewards = np.array(all_rewards, dtype=np.float32)
        mean_reward = rewards.mean()
        std_reward = rewards.std() + 1e-8
        normalized_rewards = (rewards - mean_reward) / std_reward

        # 3. Compute policy loss
        total_loss = 0.0
        for i in range(0, len(all_states), self.config.group_size):
            batch_end = min(i + self.config.group_size, len(all_states))
            batch_rewards = normalized_rewards[i:batch_end]

            # Recompute current policy log probabilities
            # In production, this would use policy.forward()
            advantages = torch.tensor(batch_rewards, dtype=torch.float32)
            surrogate_loss = -advantages.mean()
            total_loss += surrogate_loss.item()

        # 4. Backpropagation
        loss_tensor = torch.tensor(total_loss, requires_grad=True)
        self.optimizer.zero_grad()
        loss_tensor.backward()
        torch.nn.utils.clip_grad_norm_(self.policy.parameters(), 1.0)
        self.optimizer.step()

        return {
            "loss": total_loss,
            "mean_reward": float(mean_reward),
            "num_samples": len(all_rewards),
        }

    async def train(self, tasks: List[CodingTask]):
        """Complete training loop"""
        for iteration in range(self.config.num_iterations):
            print(f"\n=== Iteration {iteration + 1}/{self.config.num_iterations} ===")

            all_trajectories = []

            for task in tasks:
                # 1. Setup environment
                base_sandbox = await self.setup_environment(task)
                print(f"  Task: {task.description}")

                # 2. Fork and collect trajectories in parallel
                trajectories = await self.collect_trajectories(
                    base_sandbox, task
                )
                all_trajectories.extend(trajectories)

                # 3. Clean up base environment
                await self.env.pause_sandbox(base_sandbox)

                # 4. Evaluate results
                success_rate = sum(
                    1 for t in trajectories if t["success"]
                ) / len(trajectories) if trajectories else 0.0
                print(f"    Success rate: {success_rate:.1%}")

            # 5. GRPO update
            metrics = await self.train_step(all_trajectories)
            print(f"  Train metrics: loss={metrics['loss']:.4f}, "
                  f"reward={metrics['mean_reward']:.2f}")

            # 6. Save checkpoint
            if (iteration + 1) % 10 == 0:
                torch.save(
                    self.policy.state_dict(),
                    f"checkpoint_iter_{iteration + 1}.pt"
                )


# ============================================================
# Main Entry Point
# ============================================================

async def main():
    # Initialize AgentENV client
    client = AgentENVClient("http://localhost:8000")

    # Initialize policy model (illustration: using small model)
    from transformers import AutoModelForCausalLM, AutoTokenizer

    model_name = "Qwen/Qwen2.5-1.5B-Instruct"
    policy = AutoModelForCausalLM.from_pretrained(model_name)
    tokenizer = AutoTokenizer.from_pretrained(model_name)

    # Configure training parameters
    config = GRPOConfig(
        group_size=8,
        num_iterations=100,
        learning_rate=1e-5,
        fork_count=8,
        max_steps_per_task=20,
    )

    # Create trainer
    trainer = GRPOTrainer(
        policy_model=policy,
        tokenizer=tokenizer,
        config=config,
        env_client=client,
    )

    # Start training
    try:
        await trainer.train(BENCHMARK_TASKS)
    finally:
        await client.close()


if __name__ == "__main__":
    asyncio.run(main())

#!/usr/bin/env python3
"""AgentENV network isolation configuration example"""

import ipaddress
from typing import List, Optional

class NetworkIsolationConfig:
    """Sandbox network isolation configuration"""

    # Default denied internal CIDR list
    DEFAULT_DENIED_CIDRS = [
        "10.0.0.0/8",       # Private network
        "172.16.0.0/12",    # Private network
        "192.168.0.0/16",   # Private network
        "169.254.0.0/16",   # Link-local
        "127.0.0.0/8",      # Local loopback
    ]

    def __init__(
        self,
        allow_internet: bool = True,
        denied_cidrs: Optional[List[str]] = None,
        allowed_cidrs: Optional[List[str]] = None,
    ):
        self.allow_internet = allow_internet
        self.denied_cidrs = [
            ipaddress.ip_network(cidr)
            for cidr in (denied_cidrs or self.DEFAULT_DENIED_CIDRS)
        ]
        self.allowed_cidrs = [
            ipaddress.ip_network(cidr)
            for cidr in (allowed_cidrs or [])
        ]

    def validate_egress(self, target_ip: str, target_port: int) -> bool:
        """Validate whether egress target is allowed"""
        ip = ipaddress.ip_address(target_ip)

        # Whitelist first: if in allowed_cidrs, allow
        for network in self.allowed_cidrs:
            if ip in network:
                return True

        # Blacklist: if in denied_cidrs, deny
        for network in self.denied_cidrs:
            if ip in network:
                return False

        # Default policy: based on allow_internet
        return self.allow_internet

    def to_api_request(self) -> dict:
        """Convert to AgentENV API request format"""
        return {
            "allowInternetAccess": self.allow_internet,
            "alwaysDeniedCIDRs": [str(c) for c in self.denied_cidrs],
            "alwaysAllowedCIDRs": [str(c) for c in self.allowed_cidrs],
        }


# Create network configurations for different security levels
def create_network_config(task_type: str) -> NetworkIsolationConfig:
    """Create network isolation config based on task type"""
    configs = {
        "high_security": NetworkIsolationConfig(
            allow_internet=False,
            denied_cidrs=[
                "0.0.0.0/0",  # Completely block all network access
            ],
        ),
        "coding_agent": NetworkIsolationConfig(
            allow_internet=True,
            denied_cidrs=[
                "10.0.0.0/8",
                "172.16.0.0/12",
                "192.168.0.0/16",
                "100.64.0.0/10",
            ],
            allowed_cidrs=[
                "0.0.0.0/0",  # Allow external access (e.g., pip install)
            ],
        ),
        "web_agent": NetworkIsolationConfig(
            allow_internet=True,
            denied_cidrs=[
                "10.0.0.0/8",
                "172.16.0.0/12",
                "192.168.0.0/16",
            ],
            allowed_cidrs=[
                "0.0.0.0/0",
            ],
        ),
    }
    return configs.get(task_type, configs["coding_agent"])

# Method 1: One-click install script (single node, recommended for beginners)
curl -fsSL https://raw.githubusercontent.com/kvcache-ai/AgentENV/main/scripts/install.sh | sudo bash
sudo systemctl start aenv

# Method 2: Docker deployment
curl -fsSL https://raw.githubusercontent.com/kvcache-ai/AgentENV/main/scripts/docker-setup.sh | sudo bash
docker pull ghcr.io/kvcache-ai/aenv-server:latest
docker run -d --privileged -v /dev:/dev -p 8000:8000 ghcr.io/kvcache-ai/aenv-server:latest

# Method 3: Docker Compose (simulate multi-node cluster)
# See docker-compose.yml

# Method 4: Kubernetes (production multi-node)
# Includes Gateway, Scheduler, Node DaemonSet

# Method 5: Build from source
git clone https://github.com/kvcache-ai/AgentENV.git
cd AgentENV
cargo build --release

# Install CLI
curl -fsSL https://raw.githubusercontent.com/kvcache-ai/AgentENV/main/scripts/install-cli.sh | bash

# Authenticate
aenv auth
# AENV server URL [http://localhost:8000]: http://127.0.0.1:8000
# API key: dummy

# Pull template
aenv pull ubuntu:22.04 --name ubuntu
aenv pull docker.io/library/python:3.12 --name python312

# Start sandbox with interactive shell
aenv start ubuntu

# Start sandbox in detached mode
aenv start ubuntu --detach

# List all sandboxes
aenv ls

# Execute single command
aenv exec <sandbox-id> ls -la /

# Pause/Resume
aenv pause <sandbox-id>
aenv resume <sandbox-id>

# Set timeout
aenv timeout <sandbox-id> 600

# Delete sandbox
aenv delete <sandbox-id>

# docker-compose.yml — Multi-node AgentENV cluster
version: "3.8"

services:
  gateway:
    image: ghcr.io/kvcache-ai/aenv-gateway:latest
    ports:
      - "8080:8080"
    environment:
      - SCHEDULER_URL=http://scheduler:9090
    depends_on:
      - scheduler

  scheduler:
    image: ghcr.io/kvcache-ai/aenv-scheduler:latest
    ports:
      - "9090:9090"
    environment:
      - NODE_LABELS=region=us-east-1,zone=a

  node-1:
    image: ghcr.io/kvcache-ai/aenv-server:latest
    privileged: true
    devices:
      - /dev/kvm:/dev/kvm
    volumes:
      - /dev:/dev
      - aenv-data-1:/var/lib/aenv
    environment:
      - SCHEDULER_URL=http://scheduler:9090
      - NODE_ID=node-1
    depends_on:
      - scheduler

  node-2:
    image: ghcr.io/kvcache-ai/aenv-server:latest
    privileged: true
    devices:
      - /dev/kvm:/dev/kvm
    volumes:
      - /dev:/dev
      - aenv-data-2:/var/lib/aenv
    environment:
      - SCHEDULER_URL=http://scheduler:9090
      - NODE_ID=node-2
    depends_on:
      - scheduler

volumes:
  aenv-data-1:
  aenv-data-2:

#!/usr/bin/env python3
"""AgentENV performance benchmark script"""

import asyncio
import time
import statistics
from dataclasses import dataclass, field
from typing import List
from agentenv_client import AgentENVClient

@dataclass
class BenchmarkResult:
    """Benchmark test results"""
    cold_start_ms: List[float] = field(default_factory=list)
    snapshot_boot_ms: List[float] = field(default_factory=list)
    pause_ms: List[float] = field(default_factory=list)
    resume_ms: List[float] = field(default_factory=list)
    fork_ms: List[float] = field(default_factory=list)
    snapshot_create_ms: List[float] = field(default_factory=list)

    def summary(self) -> str:
        """Generate test summary"""
        lines = []
        lines.append("=== AgentENV Performance Benchmark ===")
        lines.append("")
        for name, data in [
            ("Cold Boot Time", self.cold_start_ms),
            ("Snapshot Boot Time", self.snapshot_boot_ms),
            ("Pause Time", self.pause_ms),
            ("Resume Time", self.resume_ms),
            ("Fork Time", self.fork_ms),
            ("Snapshot Create Time", self.snapshot_create_ms),
        ]:
            if data:
                lines.append(
                    f"{name}: "
                    f"avg={statistics.mean(data):.1f}ms, "
                    f"p50={statistics.median(data):.1f}ms, "
                    f"p99={sorted(data)[int(len(data)*0.99)]:.1f}ms, "
                    f"min={min(data):.1f}ms, max={max(data):.1f}ms"
                )
        return "\n".join(lines)

async def run_benchmark(
    client: AgentENVClient,
    iterations: int = 50,
) -> BenchmarkResult:
    """Run benchmark tests"""
    result = BenchmarkResult()

    # 1. Test cold boot
    print(f"Testing cold boot ({iterations} times)...")
    for i in range(iterations):
        start = time.perf_counter()
        sid = await client.create_sandbox(
            template="ubuntu:22.04",
            vcpu=2,
            memory_mb=1024,
        )
        elapsed = (time.perf_counter() - start) * 1000
        result.cold_start_ms.append(elapsed)
        await client.delete_sandbox(sid)

    # 2. Create a template, then test snapshot boot
    print("Creating template snapshot...")
    builder = await client.create_sandbox(
        template="ubuntu:22.04",
        vcpu=2,
        memory_mb=1024,
    )
    await client.exec_command(builder, "apt-get update && apt-get install -y python3-pip")
    template_id = await client.create_template(builder, "benchmark-template")
    await client.delete_sandbox(builder)

    # 3. Test snapshot boot
    print(f"Testing snapshot boot ({iterations} times)...")
    for i in range(iterations):
        start = time.perf_counter()
        sid = await client.create_sandbox(template=template_id)
        elapsed = (time.perf_counter() - start) * 1000
        result.snapshot_boot_ms.append(elapsed)

        # Test pause
        start = time.perf_counter()
        await client.pause_sandbox(sid)
        elapsed = (time.perf_counter() - start) * 1000
        result.pause_ms.append(elapsed)

        # Test resume
        start = time.perf_counter()
        await client.resume_sandbox(sid)
        elapsed = (time.perf_counter() - start) * 1000
        result.resume_ms.append(elapsed)

        # Test snapshot creation
        start = time.perf_counter()
        await client.create_snapshot(sid, f"snap-{i}")
        elapsed = (time.perf_counter() - start) * 1000
        result.snapshot_create_ms.append(elapsed)

        await client.delete_sandbox(sid)

    # 4. Test Fork
    print(f"Testing Fork ({min(iterations, 16)} times)...")
    parent = await client.create_sandbox(template=template_id)
    start = time.perf_counter()
    children = await client.fork_sandbox(parent, count=16)
    elapsed = (time.perf_counter() - start) * 1000
    result.fork_ms.append(elapsed)
    for child in children:
        await client.delete_sandbox(child)
    await client.delete_sandbox(parent)

    return result

async def main():
    client = AgentENVClient("http://localhost:8000")
    try:
        result = await run_benchmark(client, iterations=30)
        print(result.summary())
    finally:
        await client.close()

if __name__ == "__main__":
    asyncio.run(main())

# E2B SDK code — runs on AgentENV without modification
import os
from e2b import Sandbox

# Just change this line:
os.environ["E2B_API_URL"] = "http://localhost:8000"

# The following code remains completely unchanged
sandbox = Sandbox(template="ubuntu:22.04")
sandbox.commands.run("pip install requests")
code = """
import requests
response = requests.get('https://api.example.com/data')
print(response.json())
"""
sandbox.files.write("/tmp/script.py", code)
result = sandbox.commands.run("python /tmp/script.py")
print(result.stdout)
sandbox.close()