swarm-sandbox

name	swarm-sandbox
description	Safe isolated testing environment for multi-agent swarm topologies before production deployment

Swarm Sandbox Simulator

When to activate

• Before deploying multi-agent swarms with council.js to production
• Testing agent interaction patterns, message flows, and error handling
• Validating swarm topology configurations (2-agent, 5-agent, council-style)
• Experimenting with new agent communication protocols
• Sandboxing edge cases and failure scenarios in isolated environments
• Validating safety guardrails and rate limits before live deployment

When NOT to use

• Single-agent operations without inter-agent communication
• Production deployments — use this for testing only, then migrate to live
• Simple sequential agent pipelines — standard agent-construction skill is sufficient
• One-off quick experiments without formal validation needs
• Legacy monolithic systems without agent-based architecture

Instructions

Core concept

Swarm Sandbox creates an isolated, dry-run environment for testing multi-agent topologies. It validates:

1. Topology structure — agent count, relationships, message flow
2. Communication paths — request routing, response handling, cycles
3. Safety guardrails — rate limits, timeout handling, resource isolation
4. Error propagation — failure modes, recovery strategies, circuit breakers
5. Integration readiness — environment compatibility before production

Sandbox topology templates

2-Agent Pair (Simple Request-Response)

// Mock environment for bilateral agent communication
const sandbox2Agent = {
  agents: [
    {
      id: "agent-primary",
      role: "orchestrator",
      model: "claude-opus-4-20250514",
      capabilities: ["task_decomposition", "delegation", "synthesis"],
      timeout_ms: 30000,
      rate_limit: { requests_per_minute: 60 }
    },
    {
      id: "agent-specialist",
      role: "specialist",
      model: "claude-opus-4-20250514",
      capabilities: ["analysis", "execution"],
      timeout_ms: 30000,
      rate_limit: { requests_per_minute: 120 }
    }
  ],
  communication: {
    topology: "bilateral",
    message_protocol: "json-rpc",
    max_queue_depth: 1000,
    retry_policy: {
      max_attempts: 3,
      backoff_ms: [100, 250, 500],
      circuit_breaker: { threshold: 5, reset_ms: 60000 }
    }
  },
  isolation: {
    context_isolation: true,
    memory_limit_mb: 512,
    cpu_quota_percent: 50,
    network_restricted: true
  },
  monitoring: {
    log_level: "debug",
    trace_enabled: true,
    metrics: ["latency_ms", "error_rate", "queue_depth", "token_usage"]
  }
};

// Mock simulation environment
const mock2AgentEnv = {
  // Agent-primary sends task to agent-specialist
  scenario_1: {
    name: "Standard delegation",
    steps: [
      {
        actor: "agent-primary",
        action: "message",
        target: "agent-specialist",
        payload: {
          type: "task",
          id: "task-001",
          instruction: "Analyze market sentiment for tech stocks",
          context: { data_url: "mock://data/sentiment.json" }
        },
        expected_latency_ms: [500, 3000]
      },
      {
        actor: "agent-specialist",
        action: "response",
        target: "agent-primary",
        payload: {
          task_id: "task-001",
          status: "completed",
          result: { sentiment_score: 0.72, confidence: 0.94 }
        },
        expected_latency_ms: [300, 2000]
      }
    ]
  },
  
  // Failure mode: specialist timeout
  scenario_2: {
    name: "Timeout recovery",
    steps: [
      {
        actor: "agent-primary",
        action: "message",
        target: "agent-specialist",
        payload: {
          type: "task",
          id: "task-002",
          instruction: "Process large dataset",
          timeout_override_ms: 5000
        }
      },
      {
        actor: "agent-specialist",
        action: "timeout",
        duration_ms: 6000,
        expected_behavior: "circuit_breaker_activates"
      },
      {
        actor: "agent-primary",
        action: "retry_with_backoff",
        target: "agent-specialist",
        backoff_attempt: 1,
        delay_ms: 100
      }
    ]
  },
  
  // Failure mode: resource exhaustion
  scenario_3: {
    name: "Rate limit enforcement",
    steps: [
      {
        actor: "agent-primary",
        action: "burst_messages",
        count: 150,
        target: "agent-specialist",
        rate_limit_quota: 120
      },
      {
        actor: "sandbox",
        action: "enforce_limit",
        expected_behavior: "reject_excess_requests",
        rejected_count: 30
      }
    ]
  }
};

5-Agent Council (Hub-and-Spoke)

// Council topology with central orchestrator
const sandbox5AgentCouncil = {
  agents: [
    {
      id: "council-orchestrator",
      role: "orchestrator",
      model: "claude-opus-4-20250514",
      responsibilities: ["task_routing", "synthesis", "final_decision"]
    },
    {
      id: "researcher-agent",
      role: "researcher",
      model: "claude-opus-4-20250514",
      responsibilities: ["data_gathering", "fact_checking", "source_validation"]
    },
    {
      id: "analyst-agent",
      role: "analyst",
      model: "claude-opus-4-20250514",
      responsibilities: ["pattern_detection", "trend_analysis", "anomaly_detection"]
    },
    {
      id: "risk-agent",
      role: "risk_assessor",
      model: "claude-opus-4-20250514",
      responsibilities: ["risk_evaluation", "mitigation_planning", "compliance_check"]
    },
    {
      id: "writer-agent",
      role: "writer",
      model: "claude-opus-4-20250514",
      : [, , ]
    }
  ],
  : {
    : ,
    : ,
    : [, , , ],
    : ,
    : 
  },
  : {
    : {
      : ,
      : ,
      : [],
      : 
    },
    : {
      : ,
      : ,
      : [],
      : []
    },
    : {
      : ,
      : ,
      : [],
      : [, ]
    },
    : {
      : ,
      : ,
      : [],
      : [, ]
    }
  },
  : {
    : { : , :  },
    : { : , :  },
    : { : , :  },
    : { : , :  }
  }
};


 mock5AgentScenarios = {
  : {
    : ,
    : [
      ,
      ,
      ,
      ,
      ,
      ,
      ,
      ,
      
    ],
    : ,
    : 
  },
  
  : {
    : ,
    : { :  },
    : ,
    : ,
    : 
  },
  
  : {
    : ,
    : { :  },
    : ,
    : ,
    : 
  },
  
  : {
    : ,
    : ,
    : ,
    : ,
    : ,
    : 
  }
};

Council-Style (Peer Collaboration with Router)

// Peer-to-peer with central message router
const sandboxPeerCouncil = {
  agents: [
    {
      id: "peer-1-sdr",
      role: "sales_development",
      model: "claude-opus-4-20250514",
      peer_topics: ["lead_qualification", "outreach_strategy"]
    },
    {
      id: "peer-2-marketing",
      role: "marketing_specialist",
      model: "claude-opus-4-20250514",
      peer_topics: ["campaign_analysis", "content_strategy"]
    },
    {
      id: "peer-3-product",
      role: "product_owner",
      model: "claude-opus-4-20250514",
      peer_topics: ["feature_analysis", "roadmap_planning"]
    }
  ],
  router: {
    type: "message_router",
    routing_rules: [
      { topic: "lead_qualification", route_to: "peer-1-sdr" },
      { topic: "campaign_analysis", route_to: "peer-2-marketing" },
      { topic: "feature_analysis", route_to: "peer-3-product" },
      { topic: "cross_functional_review", route_to:  }
    ],
    : { : , :  },
    : 
  },
  : {
    : [
      { : , : , :  },
      { : , : , :  },
      { : , : , :  }
    ],
    : {
      : ,
      : ,
      : 
    }
  }
};

Mock environment variables and dry-run flags

# Sandbox configuration file: .swarm-sandbox.env
SWARM_MODE=sandbox
SANDBOX_ISOLATION_LEVEL=strict          # strict | moderate | permissive
SANDBOX_NETWORK_ACCESS=none             # none | local_only | restricted_urls
SANDBOX_TIMEOUT_OVERRIDE_MS=10000       # Override all agent timeouts for testing
SANDBOX_RATE_LIMIT_MULTIPLIER=0.5       # Reduce rate limits for stress testing

# Model configuration
SANDBOX_MODEL_PRIMARY=claude-opus-4-20250514
SANDBOX_MODEL_FALLBACK=claude-sonnet-4-20250514
SANDBOX_TOKEN_LIMIT_OVERRIDE=50000      # Dry-run token budget

# Failure injection (chaos engineering)
SANDBOX_INJECT_FAILURES=false           # Enable chaos mode
SANDBOX_FAILURE_RATE_PERCENT=10         # Random failure injection rate
SANDBOX_FAILURE_MODES=timeout,malformed_response,no_route  # Which failures to inject

# Monitoring and validation
SANDBOX_TRACE_ENABLED=true
SANDBOX_VALIDATE_SCHEMA=true
SANDBOX_COLLECT_METRICS=true
SANDBOX_METRICS_OUTPUT=./sandbox-metrics.json

# Safety guardrails
SANDBOX_MAX_CONCURRENT_AGENTS=10
SANDBOX_MAX_MESSAGE_SIZE_BYTES=1048576  # 1MB
SANDBOX_MAX_TOTAL_REQUESTS_PER_RUN=5000
SANDBOX_CIRCUIT_BREAKER_THRESHOLD=5
SANDBOX_CIRCUIT_BREAKER_RESET_MS=60000

Isolation validation checklist

## Pre-Deployment Validation Checklist

### Topology & Structure
- [ ] All agent IDs are unique and non-conflicting
- [ ] Agent role assignments align with responsibilities
- [ ] Communication topology matches documented design (hub-spoke, peer, linear)
- [ ] No circular dependencies in phase sequencing (if applicable)
- [ ] All agents have reachable endpoints in sandbox mock

### Communication & Messaging
- [ ] Message protocol defined and validated (JSON-RPC, gRPC, etc.)
- [ ] Serialization/deserialization tested for all message types
- [ ] Maximum message size limits enforced
- [ ] Routing rules tested for all agent pairs
- [ ] Cycle detection working for peer-to-peer topologies
- [ ] Message queuing and backpressure mechanisms validated

### Safety & Guardrails
- [ ] Rate limits enforced per agent and globally
- [ ] Timeout handling tested (graceful degradation vs. failure)
- [ ] Circuit breakers activate after N consecutive failures
- [ ] Memory isolation per agent sandboxed
- [ ] Resource limits (CPU, memory, token budget) enforced
- [ ] Network access restricted to mock endpoints only

### Error Handling & Recovery
- [ ] Retry logic tested with backoff (exponential, jitter)
- [ ] Fallback paths exist and are validated for critical failures
- [ ] Partial failure scenarios handled (e.g., 1 of 5 agents fails)
- [ ] Error messages logged with sufficient context
- [ ] No sensitive data in error outputs

### Performance & Scale
- [ ] Baseline latency measured for each agent
- [ ] Load test: validate behavior at 2x expected concurrent tasks
- [ ] Load test: validate behavior at 5x expected message rate
- [ ] Queue depth remains within limits under stress
 [ ] No memory leaks observed in extended runs


 [ ] All external dependencies mocked (APIs, databases, file systems)
 [ ] Production environment variables documented
 [ ] Secrets management (API keys, auth tokens) not embedded in configs
 [ ] Agent discovery mechanism (if dynamic) tested in sandbox
 [ ] Graceful shutdown sequence verified


 [ ] All agent actions logged for audit trail
 [ ] Sensitive data (user IDs, PII) masked in logs
 [ ] Audit retention policy defined and enforced
 [ ] Compliance violations detected and reported
 [ ] Sandbox run reports generated automatically

Safety guardrails and validation steps

Built-in safety mechanisms

# Safety validator for sandbox configurations
class SwarmSandboxValidator:
    """Validate swarm topology before deployment."""
    
    def __init__(self, topology_config: dict):
        self.config = topology_config
        self.errors = []
        self.warnings = []
    
    def validate_all(self) -> bool:
        """Run all validation checks."""
        self.validate_topology_structure()
        self.validate_communication_graph()
        self.validate_resource_limits()
        self.validate_safety_guardrails()
        self.validate_error_handling()
        return len(self.errors) == 0
    
    def validate_topology_structure(self):
        """Verify agent IDs are unique, roles defined, etc."""
        agent_ids = set()
        for agent in self.config.get("agents", []):
            if agent["id"] in agent_ids:
                self.errors.append(f"Duplicate agent ID: {agent['id']}")
            agent_ids.add(agent["id"])
            
            if  agent.get():
                .errors.append()
              agent.get():
                .warnings.append()
    
     ():
        
        agents_map = {a[]: a  a  .config.get(, [])}
        
        
        visited = ()
        rec_stack = ()
        
         ():
            visited.add(node)
            rec_stack.add(node)
            
             neighbor  graph.get(node, []):
                 neighbor   visited:
                     has_cycle(neighbor, graph):
                         
                 neighbor  rec_stack:
                     
            
            rec_stack.remove(node)
             
        
        
        graph = {}
         agent_id  agents_map:
            graph[agent_id] = []
        
        
         rule  .config.get(, {}).get(, []):
            from_agent = rule.get()
            to_agent = rule.get()
             from_agent  graph  to_agent  agents_map:
                graph[from_agent].append(to_agent)
        
         agent_id  graph:
             agent_id   visited:
                 has_cycle(agent_id, graph):
                    .errors.append()
    
     ():
        
         agent  .config.get(, []):
            timeout = agent.get(, )
             timeout < :
                .errors.append()
             timeout > :
                .warnings.append()
            
            memory = agent.get(, )
             memory < :
                .errors.append()
    
     ():
        
        isolation = .config.get(, {})
          isolation.get():
            .warnings.append()
          isolation.get():
            .errors.append()
    
     ():
        
        comm = .config.get(, {})
        retry = comm.get(, {})
        
          retry:
            .warnings.append()
        :
              retry.get():
                .warnings.append()
            
            max_attempts = retry.get(, )
             max_attempts < :
                .errors.append()
    
     () -> :
        
        lines = []
         .errors:
            lines.append()
             err  .errors:
                lines.append()
         .warnings:
            lines.append()
             warn  .warnings:
                lines.append()
          .errors   .warnings:
            lines.append()
         .join(lines)

Runtime validation during dry-run

# Monitor and validate swarm behavior in real-time
class SwarmMonitor:
    """Track metrics and validate swarm health during sandbox execution."""
    
    def __init__(self, config):
        self.config = config
        self.metrics = {
            "agent_latencies": {},
            "message_counts": {},
            "error_counts": {},
            "queue_depths": {},
            "circuit_breaker_trips": {}
        }
        self.alerts = []
    
    def record_message(self, from_agent: str, to_agent: str, latency_ms: float):
        """Record message delivery."""
        if from_agent not in self.metrics["message_counts"]:
            self.metrics["message_counts"][from_agent] = 0
        self.metrics["message_counts"][from_agent] += 1
        
        if to_agent not in self.metrics["agent_latencies"]:
            self.metrics["agent_latencies"][to_agent] = []
        self.metrics["agent_latencies"][to_agent].append(latency_ms)
    
    def record_error():
        
        key = 
        .metrics[][key] = .metrics[].get(key, ) + 
    
     () -> :
        
        health = {: , : []}
        
        
         agent, latencies  .metrics[].items():
             latencies:
                avg_latency = (latencies) / (latencies)
                max_latency = (latencies)
                
                expected_max = .config.get(, [{}])[].get(, )
                 max_latency > expected_max * :
                    health[] = 
                    health[].append()
        
        
        total_errors = (.metrics[].values())
        total_messages = (.metrics[].values())
         total_messages > :
            error_rate = total_errors / total_messages
             error_rate > :  
                health[] = 
                health[].append()
        
         health

Example

Complete sandbox setup and validation

#!/usr/bin/env python3
"""
Swarm Sandbox Simulator: End-to-end example.
Setup a 5-agent council, validate topology, run dry-run scenario,
and generate deployment report.
"""

import json
from pathlib import Path
from swarm_sandbox import (
    SwarmSandboxValidator,
    SwarmMonitor,
    DryRunSimulator,
    SandboxEnvironment
)

# 1. Define swarm topology
COUNCIL_TOPOLOGY = {
    "name": "customer_analysis_council",
    "agents": [
        {
            "id": "orchestrator",
            "role": "orchestrator",
            "model": "claude-opus-4-20250514",
            "timeout_ms": 30000,
            "rate_limit": {"requests_per_minute": 60}
        },
        {
            "id": "researcher",
            "role": "researcher",
            "model": "claude-opus-4-20250514",
            "timeout_ms": 20000,
            "rate_limit": {"requests_per_minute": 120}
        },
        {
            "id": "analyst",
            "role": "analyst",
            "model": "claude-opus-4-20250514",
            "timeout_ms": 20000,
            "rate_limit": {"requests_per_minute": 120}
        },
        {
            : ,
            : ,
            : ,
            : ,
            : {: }
        },
        {
            : ,
            : ,
            : ,
            : ,
            : {: }
        }
    ],
    : {
        : ,
        : ,
        : {
            : ,
            : [, , ],
            : {: , : }
        }
    },
    : {
        : ,
        : ,
        : 
    }
}


()
validator = SwarmSandboxValidator(COUNCIL_TOPOLOGY)
  validator.validate_all():
    ()
    (validator.report())
    exit()
()
(validator.report())


()
env = SandboxEnvironment(
    topology=COUNCIL_TOPOLOGY,
    isolation_level=,
    network_access=
)
()


()
monitor = SwarmMonitor(COUNCIL_TOPOLOGY)
()


()
simulator = DryRunSimulator(
    topology=COUNCIL_TOPOLOGY,
    monitor=monitor,
    sandbox_env=env
)


scenario = {
    : ,
    : ,
    : [
        {: , : , : []},
        {: , : , : },
        {: , : , : []},
        {: , : , : },
        {: , : , : []},
        {: , : , : },
        {: , : , : []},
        {: , : , : },
    ]
}

result = simulator.run_scenario(scenario)
health = monitor.validate_health()

()
()
()
()
()


()
report = {
    : COUNCIL_TOPOLOGY[],
    : {
        : ,
        : (validator.errors),
        : (validator.warnings)
    },
    : {
        : scenario[],
        : health[],
        : result[],
        : result[],
        :   result[] >   
    },
    :   health[] ==     health[] ==   ,
    : [
        ,
        ,
        ,
        ,
        
    ]
}

report_path = Path()
report_path.write_text(json.dumps(report, indent=))
()

( + *)
( + report[].upper())
(*)
()
 step  report[]:
    ()
()


exit(  report[] ==   )

Running the sandbox validator from CLI

# Initialize sandbox environment
source .swarm-sandbox.env
export SANDBOX_MODE=true

# Run topology validation
python3 swarm_sandbox_validator.py \
  --topology ./council-topology.json \
  --isolation-level strict \
  --output ./validation-report.json

# Run specific dry-run scenario
python3 swarm_sandbox_simulator.py \
  --topology ./council-topology.json \
  --scenario healthy_council \
  --metrics-output ./metrics.json \
  --trace-enabled

# Run chaos test (failure injection)
python3 swarm_sandbox_simulator.py \
  --topology ./council-topology.json \
  --chaos-mode true \
  --failure-rate-percent 10 \
  --failure-modes timeout,malformed_response \
  --output ./chaos-report.json

# Generate deployment readiness report
python3 swarm_sandbox_report.py \
  --validation ./validation-report.json \
  --metrics ./metrics.json \
  --output ./deployment-readiness.md

Integration with council.js

After sandbox validation passes, migrate to production:

// council.js controller integration
const council = require('council.js');
const sandboxReport = require('./.swarm-sandbox-report.json');

// Verify sandbox validation before deploying
if (sandboxReport.readiness !== 'green') {
  throw new Error('Sandbox validation failed. Fix issues before deployment.');
}

// Load production topology (same config as sandbox-validated)
const productionTopology = require('./council-topology.json');

// Initialize with guardrails from sandbox testing
const swarmController = council.create({
  topology: productionTopology,
  
  // Apply safety limits from sandbox
  rate_limits: productionTopology.agents.map(a => ({
    agent_id: a.id,
    requests_per_minute: a.rate_limit.requests_per_minute
  })),
  
  // Use timeout values from sandbox
  timeouts: productionTopology.agents.map(a => ({
    agent_id: a.id,
    timeout_ms: a.timeout_ms
  })),
  
  // Circuit breaker settings from sandbox
  circuit_breaker: productionTopology...,
  
  
  : {
    : ,
    : ,
    : 
  }
});


swarmController.();

Checklist: From sandbox to production

## Pre-Production Deployment Checklist

- [ ] Sandbox validation report shows "green" readiness
- [ ] All topology errors resolved (0 errors in report)
- [ ] Warnings reviewed and accepted
- [ ] Dry-run scenario executed successfully
- [ ] Chaos test (failure injection) passed with acceptable recovery
- [ ] Load test: 2x concurrency passed
- [ ] Load test: 5x message rate passed
- [ ] Production environment variables configured
- [ ] Secrets (API keys) injected at runtime
- [ ] Production database/external services ready
- [ ] Monitoring dashboards configured
- [ ] Alerting thresholds set based on sandbox metrics
- [ ] Runbook for common failure scenarios prepared
- [ ] Team trained on council operation and debugging
- [ ] Rollback plan documented (fallback to single agent)
- [ ] Start with canary: deploy to 10% of traffic first
- [ ] Monitor production metrics for first 24 hours
- [ ] Gradually increase load to 100% over 1 week