# Code Execution

Sandboxed code execution for agents: environments, security, patterns, and tools.

---

## Why Code Execution?

| Capability | Use Case |
|---|---|
| **Data analysis** | Pandas, NumPy, SQL, visualization |
| **Code generation** | Write, test, debug code |
| **Math/computation** | SymPy, numerical solving |
| **Prototyping** | Rapid iteration, REPL |
| **Verification** | Run tests, type check, lint |
| **Automation** | Scripts, file ops, API calls |

---

## Sandbox Options

| Sandbox | Type | Startup | Isolation | Best For |
|---|---|---|---|---|
| **E2B** | Cloud VM | ~1s | Strong (VM) | Production, long-running |
| **Modal** | Cloud functions | ~100ms | Strong (container) | GPU, custom images |
| **Daytona** | Cloud workspaces | ~2s | Strong (VM) | Persistent, IDE integration |
| **Firecracker** | MicroVM | ~100ms | Strongest (KVM) | High-security, multi-tenant |
| **gVisor** | User-space kernel | ~10ms | Strong (no KVM needed) | Containers, no KVM |
| **Docker** | Container | ~100ms | Medium | Standard workloads |
| **WASM** | Wasmtime/Wasmer | ~1ms | Strong (no kernel) | Edge, plugins, untrusted |
| **Restricted Python** | Interpreter | ~1ms | Weak (AST-based) | Simple, trusted code only |
| **Local (subprocess)** | Local | ~1ms | None (host) | Trusted, dev only |

---

## E2B (Cloud VM)

### Setup
```bash
pip install e2b
export E2B_API_KEY="your-key"
```

### Basic Usage
```python
from e2b import Sandbox

sandbox = Sandbox()

# Run code
result = sandbox.run_code("""
import pandas as pd
df = pd.read_csv("data.csv")
print(df.describe())
""")

print(result.stdout)
print(result.stderr)
print(result.error)

# Files
sandbox.files.write("data.csv", open("data.csv", "rb").read())
result = sandbox.run_code("print(open('data.csv').read()[:100])")
```

### Advanced
```python
# Persistent sandbox
sandbox = Sandbox(template="base", timeout=300)

# Install packages
sandbox.run_code("!pip install numpy scikit-learn -q")

# GPU support
sandbox = Sandbox(gpu=True)

# Internet access (default: true)
sandbox = Sandbox(internet_access=False)

# Commands
sandbox.commands.run("ls -la")
```

---

## Modal (Cloud Functions)

### Setup
```bash
pip install modal
modal token new
```

### Basic Usage
```python
import modal

app = modal.App("code-execution")

image = modal.Image.debian_slim().pip_install(
    "pandas", "numpy", "matplotlib", "scikit-learn"
)

@app.function(image=image, timeout=300, cpu=2, memory=4096)
def run_code(code: str) -> dict:
    # Execute in isolated environment
    import subprocess, sys, tempfile, os
    
    with tempfile.NamedTemporaryFile(mode='w', suffix='.py', delete=False) as f:
        f.write(code)
        f.flush()
        try:
            result = subprocess.run(
                [sys.executable, f.name],
                capture_output=True,
                text=True,
                timeout=60
            )
            return {
                "stdout": result.stdout,
                "stderr": result.stderr,
                "returncode": result.returncode
            }
        finally:
            os.unlink(f.name)

# Local testing
if __name__ == "__main__":
    with modal.run():
        result = run_code.remote("print('Hello from Modal!')")
        print(result)
```

### GPU/GPU
```python
@app.function(
    image=image,
    gpu="A10G",  # or "H100", "A100"
    timeout=600
)
def train_model(code: str):
    # GPU-accelerated execution
    ...
```

---

## Daytona (Cloud Workspaces)

### Setup
```bash
pip install daytona
daytona auth login
```

### Usage
```python
from daytona import Daytona, CreateWorkspaceParams

daytona = Daytona()

workspace = daytona.create(CreateWorkspaceParams(
    repository="github.com/user/repo",
    # or blank for empty
))

# Execute in workspace
execution = workspace.execute("python script.py")
print(execution.output)

# File operations
workspace.fs.write_file("script.py", "print('hello')")
workspace.fs.read_file("output.txt")

# Terminal access
terminal = workspace.terminal.create()
terminal.send("python -c 'print(1+1)'")
print(terminal.read())
```

---

## Local Sandboxes

### Docker
```python
import docker

client = docker.from_env()

def run_in_docker(code: str, image="python:3.11-slim") -> dict:
    container = client.containers.run(
        image,
        command=["python", "-c", code],
        detach=True,
        mem_limit="512m",
        cpu_quota=50000,  # 0.5 CPU
        network_mode="none",  # No network
        remove=True
    )
    
    try:
        result = container.wait(timeout=60)
        logs = container.logs(stdout=True, stderr=True).decode()
        return {"exit_code": result["StatusCode"], "logs": logs}
    except Exception as e:
        container.kill()
        return {"error": str(e)}
```

### gVisor (Stronger Isolation)
```bash
# Install gVisor
# runsc install
```

```python
def run_in_gvisor(code: str) -> dict:
    # runsc provides stronger isolation than Docker
    result = subprocess.run(
        ["runsc", "run", "--", "python", "-c", code],
        capture_output=True,
        text=True,
        timeout=60
    )
    return {"stdout": result.stdout, "stderr": result.stderr}
```

### WASM (Wasmtime)
```python
import wasmtime

# Compile Python to WASM (using wasm32-wasi target)
# Then execute in Wasmtime

engine = wasmtime.Engine()
store = wasmtime.Store(engine)
module = wasmtime.Module.from_file(engine, "python.wasm")
instance = wasmtime.Instance(store, module, [])

# Call exported functions
result = instance.exports(store)["main"]()
```

---

## Code Execution Patterns

### 1. REPL / Interactive
```python
class REPLSession:
    def __init__(self, sandbox):
        self.sandbox = sandbox
        self.namespace = {}
    
    def execute(self, code: str) -> ExecutionResult:
        # Wrap in function to capture returns
        wrapped = f"""
import sys
import io
old_stdout = sys.stdout
sys.stdout = io.StringIO()
try:
    {code}
    result = sys.stdout.getvalue()
finally:
    sys.stdout = old_stdout
print(repr(result))
"""
        return self.sandbox.run_code(wrapped)
```

### 2. Notebook-Style (Cell-Based)
```python
class NotebookSession:
    def __init__(self, sandbox):
        self.sandbox = sandbox
        self.cells = []
        self.outputs = []
    
    def add_cell(self, code: str, cell_type="code"):
        self.cells.append({"code": code, "type": cell_type})
    
    def execute_cell(self, index: int) -> CellOutput:
        cell = self.cells[index]
        # Build context from previous cells
        context = "\n".join(c["code"] for c in self.cells[:index])
        full_code = context + "\n" + cell["code"]
        result = self.sandbox.run_code(full_code)
        self.outputs.append(result)
        return result
```

### 3. Code Generation + Verification
```python
async def generate_and_verify(generator, verifier, max_attempts=3):
    for attempt in range(max_attempts):
        code = await generator.generate()
        
        # Syntax check
        try:
            compile(code, "<generated>", "exec")
        except SyntaxError as e:
            continue
        
        # Run tests
        test_result = await run_tests(code)
        if test_result.passed:
            return code
        
        # Feedback to generator
        await generator.feedback(test_result.failures)
    
    raise GenerationFailed("Max attempts exceeded")
```

---

## Security

### Threat Model
| Threat | Mitigation |
|---|---|
| **Code injection** | Sandbox isolation |
| **File system access** | Chroot, read-only mounts |
| **Network access** | Network namespace, egress filtering |
| **Resource exhaustion** | cgroups: CPU, memory, pids limits |
| **Privilege escalation** | No root, no caps, user namespaces |
| **Data exfiltration** | No network, output limits |
| **Crypto mining** | CPU limits, timeout |

### Hardening Checklist
```python
SANDBOX_CONFIG = {
    # Resource limits
    "cpu_quota": 50000,        # 0.5 CPU
    "memory_limit": "512m",
    "pids_limit": 100,
    "timeout": 60,
    
    # Filesystem
    "read_only_root": True,
    "tmpfs": ["/tmp", "/home/user"],
    "no_new_privs": True,
    
    # Network
    "network": "none",  # or "bridge" with egress filter
    
    # Capabilities
    "drop_all_caps": True,
    "add_caps": [],  # None
    
    # User
    "user": "1000:1000",  # Non-root
    "working_dir": "/home/user",
    
    # Output
    "stdout_limit": "10MB",
    "stderr_limit": "10MB"
}
```

---

## Language Support

### Python
```python
# Standard
result = sandbox.run_code("print('Hello')")

# With data science stack
image = modal.Image.debian_slim().pip_install(
    "pandas", "numpy", "matplotlib", "seaborn",
    "scikit-learn", "torch", "transformers"
)
```

### JavaScript/TypeScript
```javascript
// Node.js sandbox
const result = await sandbox.runCode(`
const fetch = require('node-fetch');
const response = await fetch('https://api.example.com/data');
console.log(await response.json());
`);
```

### SQL
```python
# DuckDB for analytical SQL
result = sandbox.run_code("""
import duckdb
conn = duckdb.connect()
result = conn.execute("SELECT * FROM 'data.parquet' LIMIT 10").fetchdf()
print(result.to_markdown())
""")
```

### R, Julia, Go, Rust
```python
# Any language with container/WASM support
# Just need appropriate base image or WASM target
```

---

## Output Handling

### Capture & Format
```python
class CodeExecutor:
    def __init__(self, sandbox):
        self.sandbox = sandbox
    
    def execute(self, code: str, capture="all") -> ExecutionResult:
        # Wrap to capture stdout/stderr/return value
        wrapped = self._wrap_for_capture(code)
        result = self.sandbox.run_code(wrapped)
        
        return ExecutionResult(
            stdout=result.stdout,
            stderr=result.stderr,
            return_value=self._parse_return(result.stdout),
            error=result.error,
            execution_time=result.execution_time
        )
    
    def _wrap_for_capture(self, code: str) -> str:
        return f"""
import sys, io, json, traceback
old_stdout, old_stderr = sys.stdout, sys.stderr
sys.stdout, sys.stderr = io.StringIO(), io.StringIO()
try:
    result = {code}
    output = {{"stdout": sys.stdout.getvalue(), "stderr": sys.stderr.getvalue(), "return": result}}
except Exception as e:
    output = {{"stdout": sys.stdout.getvalue(), "stderr": traceback.format_exc(), "return": None}}
finally:
    sys.stdout, sys.stderr = old_stdout, old_stderr
print(json.dumps(output))
"""
```

### Streaming Output
```python
async def execute_streaming(code: str) -> AsyncIterator[OutputChunk]:
    # For long-running code with progress
    process = await asyncio.create_subprocess_exec(
        "python", "-c", code,
        stdout=asyncio.subprocess.PIPE,
        stderr=asyncio.subprocess.PIPE
    )
    
    async def read_stream(stream, type_):
        async for line in stream:
            yield OutputChunk(type=type_, content=line.decode())
    
    async with asyncio.TaskGroup() as tg:
        tg.create_task(read_stream(process.stdout, "stdout"))
        tg.create_task(read_stream(process.stderr, "stderr"))
    
    await process.wait()
```

---

## Related Notes

- [[ToolUse/Function-Calling.md]] — Code execution as tool
- [[ToolUse/MCP-Protocol.md]] — Code execution via MCP
- [[ToolUse/Browser-Automation.md]] — Browser automation
- [[ToolUse/API-Integration.md]] — API calls from code
- [[ToolUse/Custom-Tools.md]] — Building execution tools
- [[Frameworks/OpenAI-Agents-SDK.md]] — Code interpreter tool
- [[Frameworks/LangGraph.md]] — Code execution in graphs
- [[Frameworks/AgentKit.md]] — Code execution in AgentKit
- [[Frameworks/PydanticAI.md]] — Code execution tools
- [[Domains/Coding-Agents.md]] — Code agents
- [[Domains/Data-Analysis-Agents.md]] — Data analysis agents
- [[Security/Secure-Coding.md]] — Sandbox security
- [[Production/Security.md]] — Sandbox security
- [[Production/Observability.md]] — Execution observability

---

*Sources: E2B docs, Modal docs, Daytona docs, Docker security, gVisor, Firecracker, Wasmtime, Python sandboxing, modal.com, e2b.dev, daytona.io, NIST sandboxing guidelines.*