Hot Reload and Development Workflow¶

# agents/dev_utils.py
import os
import time
from datetime import datetime
from functools import wraps

def development_mode(func):
    """Decorator for development-only features"""
    @wraps(func)
    def wrapper(*args, **kwargs):
        if os.getenv('MCP_DEV_MODE') == 'true':
            print(f"[DEV] {func.__name__} called at {datetime.now()}")
            start = time.time()
            result = func(*args, **kwargs)
            print(f"[DEV] {func.__name__} took {time.time() - start:.3f}s")
            return result
        return func(*args, **kwargs)
    return wrapper

# State that persists across reloads
_persistent_cache = {}

def get_persistent_cache():
    """Access cache that survives hot reloads"""
    return _persistent_cache

@tool(capability="dev_agent")
@development_mode
def process_with_cache(key: str, value: str = None):
    """Example using persistent cache"""
    cache = get_persistent_cache()

    if value is not None:
        cache[key] = value
        return f"Stored {key}={value}"

    return cache.get(key, "Not found")

# watch_config.py
import watchdog.events
import watchdog.observers

class CustomReloadHandler(watchdog.events.FileSystemEventHandler):
    """Custom handler for fine-grained reload control"""

    def __init__(self, reload_callback):
        self.reload_callback = reload_callback
        self.last_reload = 0
        self.reload_delay = 0.5  # seconds

    def should_reload(self, event):
        """Determine if file change should trigger reload"""
        # Skip temporary files
        if event.src_path.endswith('.tmp') or '~' in event.src_path:
            return False

        # Skip test files during normal development
        if 'test_' in event.src_path and not os.getenv('RELOAD_TESTS'):
            return False

        # Debounce rapid changes
        current_time = time.time()
        if current_time - self.last_reload < self.reload_delay:
            return False

        self.last_reload = current_time
        return True

    def on_modified(self, event):
        if not event.is_directory and self.should_reload(event):
            print(f"Detected change in {event.src_path}")
            self.reload_callback(event.src_path)

# At the top of your agent file
import sys
import time

# Simple circuit breaker
startup_file = '.startup_attempts'
max_attempts = 3
window_seconds = 10

try:
    with open(startup_file, 'r') as f:
        attempts = [float(line.strip()) for line in f]

    recent_attempts = [t for t in attempts if time.time() - t < window_seconds]

    if len(recent_attempts) >= max_attempts:
        print(f"Too many startup attempts ({len(recent_attempts)}) in {window_seconds}s")
        sys.exit(1)
except FileNotFoundError:
    recent_attempts = []

# Record this startup attempt
with open(startup_file, 'a') as f:
    f.write(f"{time.time()}\n")

import pickle
import atexit

STATE_FILE = '.agent_state.pkl'

def save_state():
    """Save state before reload"""
    with open(STATE_FILE, 'wb') as f:
        pickle.dump({
            'cache': cache_data,
            'counters': counters,
            'timestamp': time.time()
        }, f)

def load_state():
    """Restore state after reload"""
    try:
        with open(STATE_FILE, 'rb') as f:
            return pickle.load(f)
    except FileNotFoundError:
        return {'cache': {}, 'counters': {}}

# Register save on exit
atexit.register(save_state)

# Load on startup
state = load_state()

# tests/test_hot_reload.py
import os
import time
import tempfile
import shutil

def test_file_change_triggers_reload():
    """Test that modifying a file triggers reload"""
    with tempfile.TemporaryDirectory() as tmpdir:
        # Create test agent
        agent_file = os.path.join(tmpdir, 'test_agent.py')
        with open(agent_file, 'w') as f:
            f.write('# version 1\n')

        # Start monitoring (mock)
        changes_detected = []

        # Modify file
        time.sleep(1)
        with open(agent_file, 'a') as f:
            f.write('# version 2\n')

        # Verify reload triggered
        assert len(changes_detected) == 1

# tests/test_reload_integration.py
import subprocess
import requests
import time

def test_agent_survives_reload():
    """Test agent remains accessible after reload"""
    # Start agent with hot reload
    proc = subprocess.Popen([
        'mcp-mesh-dev', 'start', 'test_agent.py'
    ])

    time.sleep(3)

    # Verify agent is running
    response = requests.get('http://localhost:8888/health')
    assert response.status_code == 200

    # Modify agent file
    with open('test_agent.py', 'a') as f:
        f.write('\n# trigger reload\n')

    time.sleep(2)  # Wait for reload

    # Verify agent still accessible
    response = requests.get('http://localhost:8888/health')
    assert response.status_code == 200

    proc.terminate()

# Move slow imports inside functions
def process_data():
    import pandas as pd  # Import only when needed
    return pd.DataFrame()

# Cache expensive initialization
_model = None

def get_model():
    global _model
    if _model is None:
        _model = load_expensive_model()
    return _model