snakepit

# Snakepit gRPC Streaming Guide

## Overview

This guide covers Snakepit's gRPC streaming implementation - a modern replacement for stdin/stdout communication that enables real-time progress updates, progressive results, and superior performance for ML and data processing workflows.

## Table of Contents

- [Quick Start](#quick-start)
- [Installation](#installation)
- [Core Concepts](#core-concepts)
- [API Reference](#api-reference)
- [Streaming Examples](#streaming-examples)
- [Performance](#performance)
- [Migration Guide](#migration-guide)
- [Troubleshooting](#troubleshooting)

## Quick Start

### 1. Install Dependencies

```bash
# Install gRPC dependencies
make install-grpc

# Generate protocol buffer code
make proto-python

# Verify installation
python -c "import grpc, snakepit_bridge.grpc.snakepit_pb2; print('✅ gRPC ready')"
```

### 2. Basic Configuration

```elixir
# Configure Snakepit with gRPC adapter
Application.put_env(:snakepit, :adapter_module, Snakepit.Adapters.GRPCPython)
Application.put_env(:snakepit, :grpc_config, %{
  base_port: 50051,
  port_range: 100  # Uses ports 50051-50151
})

{:ok, _} = Application.ensure_all_started(:snakepit)
```

### 3. Run Your First Stream

```elixir
# Traditional (still works)
{:ok, result} = Snakepit.execute("ping", %{})

# NEW: Streaming execution
Snakepit.execute_stream("ping_stream", %{count: 5}, fn chunk ->
  IO.puts("Received: #{chunk["message"]}")
end)
```

## Installation

### Prerequisites

- **Elixir**: 1.18+ with OTP 27+
- **Python**: 3.8+ 
- **System**: `protoc` compiler (for development)

### Elixir Dependencies

```elixir
# mix.exs
def deps do
  [
    {:grpc, "~> 0.8"},
    {:protobuf, "~> 0.12"},
    # ... existing deps
  ]
end
```

### Python Dependencies

```bash
# Option 1: Install with gRPC support
cd priv/python && pip install -e ".[grpc]"

# Option 2: Install manually
pip install grpcio>=1.50.0 protobuf>=4.0.0 grpcio-tools>=1.50.0

# Option 3: All features
pip install -e ".[all]"  # Includes gRPC + MessagePack
```

### Development Setup

```bash
# Complete development environment
make dev-setup

# Or step by step:
make install-grpc      # Install Python gRPC deps
make proto-python      # Generate protobuf code
make test             # Verify everything works
```

## Core Concepts

### Communication Protocols Comparison

| Protocol | Streaming | Multiplexing | Binary Data | Setup Complexity |
|----------|-----------|--------------|-------------|------------------|
| **stdin/stdout** | ❌ | ❌ | Base64 | Simple |
| **MessagePack** | ❌ | ❌ | ✅ Native | Simple |
| **gRPC** | ✅ Native | ✅ HTTP/2 | ✅ Native | Moderate |

### Protocol Selection

```elixir
# Snakepit automatically chooses the best available protocol:

# 1. If gRPC adapter configured and available → gRPC
# 2. If MessagePack adapter configured → MessagePack  
# 3. Fallback → JSON over stdin/stdout

# Force specific protocol:
Application.put_env(:snakepit, :adapter_module, Snakepit.Adapters.GRPCPython)        # gRPC
Application.put_env(:snakepit, :adapter_module, Snakepit.Adapters.GenericPythonMsgpack)  # MessagePack
Application.put_env(:snakepit, :adapter_module, Snakepit.Adapters.GenericPythonV2)     # JSON
```

### Worker Architecture

```
┌─────────────────┐    ┌─────────────────┐    ┌─────────────────┐
│   Elixir App    │    │  gRPC Worker    │    │ Python Process  │
│                 │◄──►│   (GenServer)   │◄──►│  (gRPC Server)  │
│ execute_stream  │    │                 │    │                 │
└─────────────────┘    └─────────────────┘    └─────────────────┘
         │                       │                       │
         │              HTTP/2 Connection               │
         │                       │                       │
    Callback Handler      Connection Mgmt          Stream Handlers
```

### Session Management

```elixir
# Sessions work seamlessly with gRPC
session_id = "ml_training_#{user_id}"

# Session-based execution (maintains state)
{:ok, result} = Snakepit.execute_in_session(session_id, "initialize_model", model_config)

# Session-based streaming
Snakepit.execute_in_session_stream(session_id, "train_model", training_data, fn chunk ->
  IO.puts("Epoch #{chunk["epoch"]}: loss=#{chunk["loss"]}")
end)
```

## API Reference

### Basic Execution

```elixir
# Snakepit.execute/3 - Same as always
{:ok, result} = Snakepit.execute(command, args, timeout \\ 30_000)

# Examples
{:ok, _} = Snakepit.execute("ping", %{})
{:ok, result} = Snakepit.execute("compute", %{operation: "add", a: 5, b: 3})
```

### Streaming Execution

```elixir
# Snakepit.execute_stream/4 - NEW!
:ok = Snakepit.execute_stream(command, args, callback_fn, timeout \\ 300_000)

# Callback function receives each chunk
callback_fn = fn chunk ->
  if chunk["is_final"] do
    IO.puts("Stream complete!")
  else
    IO.puts("Progress: #{chunk["progress"]}%")
  end
end
```

### Session-based APIs

```elixir
# Session execution (existing)
{:ok, result} = Snakepit.execute_in_session(session_id, command, args, timeout \\ 30_000)

# Session streaming (NEW!)
:ok = Snakepit.execute_in_session_stream(session_id, command, args, callback_fn, timeout \\ 300_000)
```

### Error Handling

```elixir
case Snakepit.execute_stream("long_process", %{data: large_data}, callback) do
  :ok ->
    IO.puts("Stream completed successfully")
    
  {:error, :worker_timeout} ->
    IO.puts("Operation timed out")
    
  {:error, :grpc_unavailable} ->
    IO.puts("gRPC not available, check setup")
    
  {:error, reason} ->
    IO.puts("Stream failed: #{inspect(reason)}")
end
```

## Streaming Examples

### 1. ML Batch Inference

Stream inference results as each item completes:

```elixir
batch_items = ["image_001.jpg", "image_002.jpg", "image_003.jpg"]

Snakepit.execute_stream("batch_inference", %{
  model_path: "/models/resnet50.pkl",
  batch_items: batch_items
}, fn chunk ->
  if chunk["is_final"] do
    IO.puts("🎉 Batch inference complete!")
  else
    item = chunk["item"]
    prediction = chunk["prediction"] 
    confidence = chunk["confidence"]
    IO.puts("🧠 #{item}: #{prediction} (#{confidence}% confidence)")
  end
end)

# Output:
# 🧠 image_001.jpg: cat (94% confidence)
# 🧠 image_002.jpg: dog (87% confidence)  
# 🧠 image_003.jpg: bird (91% confidence)
# 🎉 Batch inference complete!
```

### 2. Large Dataset Processing

Process huge datasets with real-time progress:

```elixir
Snakepit.execute_stream("process_large_dataset", %{
  file_path: "/data/sales_data_10gb.csv",
  chunk_size: 10_000,
  operations: ["clean", "transform", "aggregate"]
}, fn chunk ->
  if chunk["is_final"] do
    stats = chunk["final_stats"]
    IO.puts("✅ Processing complete!")
    IO.puts("📊 Final stats: #{inspect(stats)}")
  else
    progress = chunk["progress_percent"]
    processed = chunk["processed_rows"]
    total = chunk["total_rows"]
    memory_mb = chunk["memory_usage_mb"]
    
    IO.puts("📊 Progress: #{progress}% (#{processed}/#{total} rows) - Memory: #{memory_mb}MB")
  end
end)

# Output:
# 📊 Progress: 10.0% (100000/1000000 rows) - Memory: 45MB
# 📊 Progress: 20.0% (200000/1000000 rows) - Memory: 47MB
# 📊 Progress: 30.0% (300000/1000000 rows) - Memory: 46MB
# ...
# ✅ Processing complete!
# 📊 Final stats: %{errors: 12, processed: 988000, skipped: 12}
```

### 3. Real-time Log Analysis

Analyze logs in real-time as new entries arrive:

```elixir
Snakepit.execute_stream("tail_and_analyze", %{
  log_path: "/var/log/app.log",
  patterns: ["ERROR", "FATAL", "OutOfMemoryError"],
  context_lines: 3
}, fn chunk ->
  if chunk["is_final"] do
    IO.puts("📊 Log analysis session ended")
  else
    severity = chunk["severity"]
    timestamp = chunk["timestamp"]
    line = chunk["log_line"]
    pattern = chunk["pattern_matched"]
    
    emoji = case severity do
      "FATAL" -> "💀"
      "ERROR" -> "🚨" 
      "WARN" -> "⚠️"
      _ -> "ℹ️"
    end
    
    IO.puts("#{emoji} [#{timestamp}] #{pattern}: #{String.slice(line, 0, 80)}...")
  end
end)

# Output:
# 🚨 [2025-01-20T15:30:45] ERROR: Database connection failed: timeout after 30s...
# ⚠️ [2025-01-20T15:30:50] WARN: High memory usage detected: 85% of heap used...
# 💀 [2025-01-20T15:31:00] FATAL: OutOfMemoryError: Java heap space exceeded...
```

### 4. Distributed Training Monitoring

Monitor ML training progress across multiple nodes:

```elixir
Snakepit.execute_stream("distributed_training", %{
  model_config: %{
    architecture: "transformer",
    layers: 24,
    hidden_size: 1024
  },
  training_config: %{
    epochs: 100,
    batch_size: 32,
    learning_rate: 0.001
  },
  dataset_path: "/data/training_set"
}, fn chunk ->
  if chunk["is_final"] do
    model_path = chunk["final_model_path"] 
    best_acc = chunk["best_val_accuracy"]
    IO.puts("🎯 Training complete! Best accuracy: #{best_acc}")
    IO.puts("💾 Model saved: #{model_path}")
  else
    epoch = chunk["epoch"]
    train_loss = chunk["train_loss"]
    val_loss = chunk["val_loss"] 
    train_acc = chunk["train_acc"]
    val_acc = chunk["val_acc"]
    lr = chunk["learning_rate"]
    
    IO.puts("📈 Epoch #{epoch}/100:")
    IO.puts("   Train: loss=#{train_loss}, acc=#{train_acc}")
    IO.puts("   Val:   loss=#{val_loss}, acc=#{val_acc}")
    IO.puts("   LR: #{lr}")
  end
end)

# Output:
# 📈 Epoch 1/100:
#    Train: loss=2.45, acc=0.12
#    Val:   loss=2.38, acc=0.15
#    LR: 0.001
# 📈 Epoch 2/100:
#    Train: loss=2.12, acc=0.23
#    Val:   loss=2.05, acc=0.28
#    LR: 0.001
# ...
# 🎯 Training complete! Best accuracy: 0.94
# 💾 Model saved: /models/transformer_best_20250120.pkl
```

### 5. Financial Data Pipeline

Process real-time financial data with technical analysis:

```elixir
Snakepit.execute_stream("realtime_stock_analysis", %{
  symbols: ["AAPL", "GOOGL", "MSFT", "TSLA"],
  indicators: ["RSI", "MACD", "SMA_20", "SMA_50"],
  alert_thresholds: %{
    rsi_oversold: 30,
    rsi_overbought: 70,
    volume_spike: 2.0
  }
}, fn chunk ->
  if chunk["is_final"] do
    IO.puts("📊 Market session ended")
  else
    symbol = chunk["symbol"]
    price = chunk["price"]
    volume = chunk["volume"] 
    rsi = chunk["rsi"]
    signal = chunk["trading_signal"]
    
    case signal do
      "BUY" -> IO.puts("🟢 #{symbol}: $#{price} - BUY signal (RSI: #{rsi})")
      "SELL" -> IO.puts("🔴 #{symbol}: $#{price} - SELL signal (RSI: #{rsi})")
      _ -> IO.puts("⚪ #{symbol}: $#{price} - HOLD (RSI: #{rsi})")
    end
  end
end)

# Output:
# ⚪ AAPL: $175.50 - HOLD (RSI: 45.2)
# 🟢 GOOGL: $142.80 - BUY signal (RSI: 28.5)
# ⚪ MSFT: $380.20 - HOLD (RSI: 52.1) 
# 🔴 TSLA: $210.15 - SELL signal (RSI: 73.8)
```

## Performance

### Streaming vs Traditional Comparison

| Metric | stdin/stdout | MessagePack | gRPC Streaming |
|--------|-------------|-------------|----------------|
| **Latency (first result)** | 5-10s | 3-7s | **0.1-0.5s** |
| **Memory usage** | Grows with result | Grows with result | **Constant** |
| **Progress visibility** | None | None | **Real-time** |
| **Cancellation** | Kill process | Kill process | **Graceful** |
| **Error granularity** | End only | End only | **Per chunk** |
| **User experience** | "Is it working?" | "Is it working?" | **Live updates** |

### Binary Serialization Performance

gRPC mode includes automatic binary serialization for large data:

| Data Size | JSON Encoding | Binary Encoding | Speedup |
|-----------|--------------|-----------------|---------|
| 1KB | 0.1ms | 0.1ms | 1x (uses JSON) |
| 10KB | 2ms | 2ms | 1x (threshold) |
| 100KB | 25ms | 3ms | **8x faster** |
| 1MB | 300ms | 30ms | **10x faster** |
| 10MB | 3500ms | 250ms | **14x faster** |

The 10KB threshold ensures optimal performance:
- Small data remains human-readable (JSON)
- Large data gets maximum performance (binary)

### Benchmarks

```bash
# Large dataset processing (1GB CSV file)
Traditional: Submit → Wait 10 minutes → Get result or timeout
gRPC Stream: Submit → Progress every 30s → Complete with stats

# ML batch inference (1000 images)  
Traditional: Submit → Wait 5 minutes → Get all predictions
gRPC Stream: Submit → Get prediction 1 → Get prediction 2 → ...

# Real-time monitoring
Traditional: Not possible
gRPC Stream: Live tail of logs/metrics with analysis
```

### Memory Usage Patterns

```elixir
# Traditional: Memory grows with result size
results = []  # Starts empty
# ... processing 1GB of data ...
results = [huge_dataset_results]  # Now using 1GB+ memory

# gRPC Streaming: Constant memory usage
Snakepit.execute_stream("process_1gb_dataset", %{}, fn chunk ->
  process_chunk(chunk)  # Handle immediately
  # chunk gets garbage collected
end)
# Memory usage stays constant regardless of total data size
```

## Migration Guide

### From stdin/stdout to gRPC

**Step 1: Install gRPC**
```bash
make install-grpc
make proto-python
```

**Step 2: Update Configuration**
```elixir
# Before
Application.put_env(:snakepit, :adapter_module, Snakepit.Adapters.GenericPythonV2)

# After  
Application.put_env(:snakepit, :adapter_module, Snakepit.Adapters.GRPCPython)
Application.put_env(:snakepit, :grpc_config, %{base_port: 50051, port_range: 100})
```

**Step 3: Keep Existing Code**
```elixir
# This still works exactly the same
{:ok, result} = Snakepit.execute("compute", %{operation: "add", a: 5, b: 3})
```

**Step 4: Add Streaming Where Beneficial**
```elixir
# Replace long-running operations with streaming versions
# Before: 
{:ok, results} = Snakepit.execute("batch_process", large_dataset)  # Blocks for minutes

# After:
Snakepit.execute_stream("batch_process", large_dataset, fn chunk ->
  update_progress_bar(chunk["progress"])
  save_partial_result(chunk["data"])
end)
```

### From MessagePack to gRPC

```elixir
# MessagePack config (keep as fallback)
Application.put_env(:snakepit, :adapters, [
  Snakepit.Adapters.GRPCPython,        # Try gRPC first
  Snakepit.Adapters.GenericPythonMsgpack  # Fallback to MessagePack
])
```

### Gradual Migration Strategy

1. **Phase 1**: Install gRPC alongside existing adapter
2. **Phase 2**: Test gRPC with non-critical workloads  
3. **Phase 3**: Move long-running operations to streaming
4. **Phase 4**: Default to gRPC, keep MessagePack as fallback
5. **Phase 5**: Full gRPC adoption

## Creating Streaming Commands in Python

This section shows how to implement custom streaming commands in your Python adapters.

### Basic Streaming Command

```python
from snakepit_bridge.base_adapter import BaseAdapter, tool

class MyAdapter(BaseAdapter):
    @tool(description="Stream progress updates", supports_streaming=True)
    def progress_stream(self, count: int = 10, delay: float = 0.5):
        """
        Stream progress updates to the client.

        For streaming tools, return a generator that yields dictionaries.
        Each yielded dict becomes a chunk sent to the Elixir callback.
        """
        import time

        for i in range(count):
            # Yield progress chunk
            yield {
                "progress": i + 1,
                "total": count,
                "percent": ((i + 1) / count) * 100,
                "message": f"Processing item {i + 1} of {count}",
                "is_final": False
            }
            time.sleep(delay)

        # Final chunk
        yield {
            "is_final": True,
            "total_processed": count,
            "message": "Stream complete"
        }
```

### ML Batch Inference Streaming

```python
from snakepit_bridge.base_adapter import BaseAdapter, tool
import time

class MLAdapter(BaseAdapter):
    def __init__(self):
        super().__init__()
        # Load your model here
        self.model = self._load_model()

    @tool(description="Batch inference with streaming results", supports_streaming=True)
    def batch_inference(self, items: list, model_path: str = None):
        """
        Process items in a batch, streaming each result as it completes.
        """
        total = len(items)

        for idx, item in enumerate(items):
            # Perform inference
            prediction, confidence = self._predict(item)

            # Yield result for this item
            yield {
                "item": item,
                "index": idx,
                "prediction": prediction,
                "confidence": confidence,
                "progress": idx + 1,
                "total": total,
                "is_final": False
            }

        # Final summary
        yield {
            "is_final": True,
            "total_processed": total,
            "message": f"Processed {total} items successfully"
        }

    def _predict(self, item):
        # Your ML inference logic here
        # This is a placeholder
        time.sleep(0.1)
        return "predicted_class", 0.95

    def _load_model(self):
        # Load your model
        return None
```

### Large Dataset Processing

```python
from snakepit_bridge.base_adapter import BaseAdapter, tool
import pandas as pd

class DataAdapter(BaseAdapter):
    @tool(description="Process large datasets with progress", supports_streaming=True)
    def process_large_dataset(self, file_path: str, chunk_size: int = 10000):
        """
        Process a large CSV file in chunks, streaming progress.
        """
        # Get total rows (for progress calculation)
        total_rows = sum(1 for _ in open(file_path)) - 1  # -1 for header

        processed = 0
        errors = 0

        # Process in chunks
        for chunk in pd.read_csv(file_path, chunksize=chunk_size):
            # Process this chunk
            try:
                results = self._process_chunk(chunk)
                processed += len(chunk)

                # Yield progress
                yield {
                    "processed_rows": processed,
                    "total_rows": total_rows,
                    "progress_percent": (processed / total_rows) * 100,
                    "memory_usage_mb": chunk.memory_usage(deep=True).sum() / 1024 / 1024,
                    "is_final": False
                }
            except Exception as e:
                errors += 1
                yield {
                    "error": str(e),
                    "chunk_start": processed,
                    "is_final": False
                }

        # Final statistics
        yield {
            "is_final": True,
            "final_stats": {
                "total_rows": total_rows,
                "processed": processed,
                "errors": errors,
                "skipped": total_rows - processed
            }
        }

    def _process_chunk(self, chunk):
        # Your processing logic
        return chunk.apply(lambda row: row, axis=1)
```

### Real-time Log Analysis

```python
from snakepit_bridge.base_adapter import BaseAdapter, tool
import re
import time

class LogAdapter(BaseAdapter):
    @tool(description="Tail and analyze logs in real-time", supports_streaming=True)
    def tail_and_analyze(self, log_path: str, patterns: list, context_lines: int = 3):
        """
        Stream log analysis results in real-time.
        """
        pattern_regexes = [re.compile(p) for p in patterns]

        # Simulate tailing (in production, use actual file watching)
        with open(log_path, 'r') as f:
            for line_num, line in enumerate(f):
                for pattern_idx, regex in enumerate(pattern_regexes):
                    if regex.search(line):
                        # Found a match
                        severity = self._determine_severity(line)

                        yield {
                            "line_number": line_num,
                            "log_line": line.strip(),
                            "pattern_matched": patterns[pattern_idx],
                            "severity": severity,
                            "timestamp": self._extract_timestamp(line),
                            "is_final": False
                        }

                # Simulate real-time (remove for actual file watching)
                time.sleep(0.01)

        yield {
            "is_final": True,
            "message": "Log analysis complete"
        }

    def _determine_severity(self, line):
        if "FATAL" in line or "CRITICAL" in line:
            return "FATAL"
        elif "ERROR" in line:
            return "ERROR"
        elif "WARN" in line:
            return "WARN"
        return "INFO"

    def _extract_timestamp(self, line):
        # Extract timestamp from log line
        # This is a placeholder
        return "2025-01-20T15:30:45"
```

### ML Training with Epoch Updates

```python
from snakepit_bridge.base_adapter import BaseAdapter, tool
import time

class TrainingAdapter(BaseAdapter):
    @tool(description="Train model with streaming epoch updates", supports_streaming=True)
    def distributed_training(self, model_config: dict, training_config: dict, dataset_path: str):
        """
        Train a model and stream updates for each epoch.
        """
        epochs = training_config.get("epochs", 100)
        best_val_acc = 0.0

        for epoch in range(1, epochs + 1):
            # Simulate training epoch
            train_loss, train_acc = self._train_epoch(epoch)
            val_loss, val_acc = self._validate_epoch(epoch)

            # Track best accuracy
            if val_acc > best_val_acc:
                best_val_acc = val_acc
                model_path = f"/models/model_epoch_{epoch}.pkl"
                self._save_checkpoint(model_path)

            # Yield epoch results
            yield {
                "epoch": epoch,
                "total_epochs": epochs,
                "train_loss": train_loss,
                "train_acc": train_acc,
                "val_loss": val_loss,
                "val_acc": val_acc,
                "learning_rate": self._get_current_lr(epoch),
                "is_final": False
            }

        # Final result
        yield {
            "is_final": True,
            "final_model_path": model_path,
            "best_val_accuracy": best_val_acc,
            "total_epochs": epochs,
            "message": "Training complete"
        }

    def _train_epoch(self, epoch):
        # Training logic (placeholder)
        time.sleep(0.1)
        return 2.5 - (epoch * 0.02), 0.1 + (epoch * 0.008)

    def _validate_epoch(self, epoch):
        # Validation logic (placeholder)
        time.sleep(0.05)
        return 2.3 - (epoch * 0.018), 0.15 + (epoch * 0.007)

    def _get_current_lr(self, epoch):
        # Learning rate schedule
        return 0.001 * (0.95 ** epoch)

    def _save_checkpoint(self, path):
        # Save model checkpoint
        pass
```

### Key Patterns for Streaming Commands

1. **Use Generator Functions**: Return a generator (using `yield`) instead of returning a value directly
2. **Mark with supports_streaming**: Add `supports_streaming=True` to the `@tool` decorator
3. **Include is_final**: Always include `"is_final": False` in intermediate chunks and `"is_final": True` in the final chunk
4. **Yield Dictionaries**: Each yielded value should be a dictionary that will be sent to the Elixir callback
5. **Progress Information**: Include progress indicators (`progress`, `total`, `percent`, etc.) to help users track completion
6. **Error Handling**: Yield error information as chunks rather than raising exceptions

### Calling Streaming Commands from Elixir

```elixir
# Call your custom streaming command
Snakepit.execute_stream("progress_stream", %{count: 10, delay: 0.5}, fn chunk ->
  if chunk["is_final"] do
    IO.puts("✅ #{chunk["message"]}")
  else
    IO.puts("📊 Progress: #{chunk["percent"]}% - #{chunk["message"]}")
  end
end)

# With session affinity
Snakepit.execute_in_session_stream("ml_session", "distributed_training", %{
  model_config: %{layers: 12},
  training_config: %{epochs: 100},
  dataset_path: "/data/train.csv"
}, fn chunk ->
  if chunk["is_final"] do
    IO.puts("🎯 Training complete! Model: #{chunk["final_model_path"]}")
  else
    IO.puts("📈 Epoch #{chunk["epoch"]}: loss=#{chunk["train_loss"]}, acc=#{chunk["train_acc"]}")
  end
end)
```

## Troubleshooting

### Common Issues

#### 1. "Generated gRPC code not found"

```bash
# Solution: Generate protobuf code
make proto-python

# Verify:
ls priv/python/snakepit_bridge/grpc/
# Should show: snakepit_pb2.py, snakepit_pb2_grpc.py
```

#### 2. "gRPC dependencies not available"

```bash
# Check Python dependencies
python -c "import grpc; print('gRPC:', grpc.__version__)"
python -c "import google.protobuf; print('Protobuf OK')"

# Install if missing
pip install 'snakepit-bridge[grpc]'
```

#### 3. "Port already in use"

```elixir
# Solution: Configure different port range
Application.put_env(:snakepit, :grpc_config, %{
  base_port: 50100,  # Try different base port
  port_range: 50     # Smaller range
})
```

#### 4. Worker initialization timeout

```bash
# Check Python process
ps aux | grep grpc_bridge

# Check Python errors
python priv/python/grpc_bridge.py --help

# Increase timeout
Application.put_env(:snakepit, :worker_init_timeout, 30_000)
```

#### 5. Streaming callback errors

```elixir
# Bad: Callback crashes on unexpected data
callback = fn chunk ->
  result = chunk["result"]  # May be nil
  process_result(result)    # Crashes if nil
end

# Good: Defensive callback
callback = fn chunk ->
  case chunk do
    %{"is_final" => true} ->
      IO.puts("Stream complete")
    %{"result" => result} when not is_nil(result) ->
      process_result(result)
    %{"error" => error} ->
      IO.puts("Stream error: #{error}")
    other ->
      IO.puts("Unexpected chunk: #{inspect(other)}")
  end
end
```

### Debug Mode

```elixir
# Enable debug logging
Logger.configure(level: :debug)

# Trace gRPC worker
:sys.trace(worker_pid, true)

# Check gRPC connection health
Snakepit.GRPCWorker.get_health(worker)
Snakepit.GRPCWorker.get_info(worker)
```

### Performance Tuning

```elixir
# Adjust gRPC settings
Application.put_env(:snakepit, :grpc_config, %{
  base_port: 50051,
  port_range: 100,
  # Increase concurrent streams
  max_concurrent_streams: 10,
  # Tune timeouts
  connection_timeout: 10_000,
  stream_timeout: 300_000
})

# Pool optimization
Application.put_env(:snakepit, :pool_config, %{
  pool_size: 16,  # More workers for concurrent streams
  worker_init_timeout: 30_000
})
```

### Fallback Strategy

```elixir
# Graceful degradation if gRPC unavailable
defmodule MyApp.SnakepitClient do
  def execute_with_fallback(command, args, opts \\ []) do
    case Snakepit.execute(command, args, opts) do
      {:ok, result} -> 
        {:ok, result}
      {:error, :grpc_unavailable} ->
        # Fallback to MessagePack
        with_adapter(Snakepit.Adapters.GenericPythonMsgpack, fn ->
          Snakepit.execute(command, args, opts)
        end)
      {:error, reason} ->
        {:error, reason}
    end
  end
  
  defp with_adapter(adapter, fun) do
    old_adapter = Application.get_env(:snakepit, :adapter_module)
    Application.put_env(:snakepit, :adapter_module, adapter)
    
    try do
      fun.()
    after
      Application.put_env(:snakepit, :adapter_module, old_adapter)
    end
  end
end
```

---