# Safety & Coordination

Ensuring multi-agent systems operate safely: containment, monitoring, guardrails, and failure modes.

---

## Safety Challenges in Multi-Agent Systems

| Challenge | Single Agent | Multi-Agent |
|---|---|---|
| **Action Attribution** | Clear | Distributed |
| **Cascading Failures** | Local | Systemic |
| **Adversarial Coordination** | Single attacker | Colluding agents |
| **Emergent Behavior** | Predictable | Unpredictable |
| **Accountability** | Single entity | Distributed |
| **Rollback** | Single state | Distributed state |

---

## Safety Layers

### 1. Individual Agent Guardrails
```python
class AgentGuardrails:
    def __init__(self, agent: Agent):
        self.agent = agent
    
    async def check_action(self, action: Action) -> GuardrailResult:
        checks = [
            self.check_permissions(action),
            self.check_resource_limits(action),
            self.check_side_effects(action),
            self.check_policy_compliance(action),
            self.check_rate_limits(action)
        ]
        
        results = await asyncio.gather(*checks)
        return GuardrailResult(
            allowed=all(r.allowed for r in results),
            violations=[r for r in results if not r.allowed]
        )
    
    async def check_permissions(self, action: Action) -> GuardrailCheck:
        required = action.required_permissions
        granted = self.agent.permissions
        missing = required - granted
        return GuardrailCheck(
            allowed=len(missing) == 0,
            violation=f"Missing permissions: {missing}" if missing else None
        )
    
    async def check_resource_limits(self, action: Action) -> GuardrailCheck:
        limits = self.agent.resource_limits
        estimated = action.estimate_resources()
        
        violations = []
        for resource, limit in limits.items():
            if estimated.get(resource, 0) > limit:
                violations.append(f"{resource}: {estimated[resource]} > {limit}")
        
        return GuardrailCheck(
            allowed=len(violations) == 0,
            violation="; ".join(violations) if violations else None
        )
```

### 2. Inter-Agent Safety Protocols

#### Action Authorization
```python
class InterAgentAuth:
    def __init__(self, policy_engine: PolicyEngine):
        self.policy = policy_engine
    
    async def authorize(self, 
                        from_agent: str, 
                        to_agent: str, 
                        action: Action) -> AuthResult:
        # Check if from_agent can request this of to_agent
        allowed = await self.policy.check(
            principal=from_agent,
            resource=to_agent,
            action=action.type,
            context=action.context
        )
        
        return AuthResult(
            allowed=allowed.allowed,
            reason=allowed.reason,
            required_approvals=allowed.required_approvals
        )
```

#### Resource Quotas
```python
class ResourceQuotaManager:
    def __init__(self, quotas: dict[str, ResourceQuota]):
        self.quotas = quotas
        self.usage: dict[str, ResourceUsage] = defaultdict(ResourceUsage)
    
    async def check_and_reserve(self, agent_id: str, request: ResourceRequest) -> bool:
        quota = self.quotas.get(agent_id)
        if not quota:
            return True  # No quota = unlimited
        
        projected = self.usage[agent_id] + request
        
        for resource, limit in quota.limits.items():
            if projected.get(resource, 0) > limit:
                return False
        
        # Reserve
        for resource, amount in request.items():
            self.usage[agent_id][resource] += amount
        return True
    
    async def release(self, agent_id: str, release: ResourceRequest):
        for resource, amount in release.items():
            self.usage[agent_id][resource] = max(0, 
                self.usage[agent_id][resource] - amount)
```

### 3. System-Level Safety

#### Circuit Breakers
```python
class SystemCircuitBreaker:
    def __init__(self, failure_threshold: int = 5, 
                 timeout: int = 60, half_open_requests: int = 3):
        self.failure_threshold = failure_threshold
        self.timeout = timeout
        self.half_open_requests = half_open_requests
        
        self.state = "closed"
        self.failure_count = 0
        self.last_failure_time = 0
        self.half_open_successes = 0
    
    async def call(self, func: Callable, *args, **kwargs):
        if self.state == "open":
            if time.time() - self.last_failure_time > self.timeout:
                self.state = "half_open"
                self.half_open_successes = 0
            else:
                raise CircuitOpenError()
        
        try:
            result = await func()
            self.on_success()
            return result
        except Exception as e:
            self.on_failure()
            raise
    
    def on_success(self):
        self.failure_count = 0
        if self.state == "half_open":
            self.half_open_successes += 1
            if self.half_open_successes >= self.half_open_requests:
                self.state = "closed"
    
    def on_failure(self):
        self.failure_count += 1
        self.last_failure_time = time.time()
        if self.failure_count >= self.failure_threshold:
            self.state = "open"
```

#### Global Rate Limiting
```python
class GlobalRateLimiter:
    def __init__(self, max_requests_per_second: int = 100):
        self.max_rps = max_requests_per_second
        self.tokens = max_requests_per_second
        self.last_refill = time.time()
        self.lock = asyncio.Lock()
    
    async def acquire(self, tokens: int = 1) -> bool:
        async with self.lock:
            now = time.time()
            # Refill
            elapsed = now - self.last_refill
            self.tokens = min(self.max_rps, self.tokens + elapsed * self.max_rps)
            self.last_refill = now
            
            if self.tokens >= tokens:
                self.tokens -= tokens
                return True
            return False
    
    async def wait_for_tokens(self, tokens: int = 1):
        while not await self.acquire(tokens):
            await asyncio.sleep(0.1)
```

### 4. Containment & Isolation

#### Sandbox per Agent
```python
class AgentSandbox:
    def __init__(self, agent_id: str, config: SandboxConfig):
        self.agent_id = agent_id
        self.config = config
        self.sandbox = self._create_sandbox()
    
    def _create_sandbox(self) -> Sandbox:
        return Sandbox(
            cpu_limit=self.config.cpu_cores,
            memory_limit=self.config.memory_mb,
            disk_limit=self.config.disk_mb,
            network=self.config.network_access,
            allowed_domains=self.config.allowed_domains,
            blocked_commands=self.config.blocked_commands,
            read_only_paths=self.config.read_only_paths,
            read_write_paths=self.config.read_write_paths
        )
    
    async def execute(self, code: str) -> ExecutionResult:
        return await self.sandbox.run(code)
    
    async def execute_tool(self, tool: Tool, args: dict) -> ToolResult:
        # Validate tool against sandbox policy
        if not self._tool_allowed(tool):
            raise ToolNotAllowed(f"Tool {tool.name} not permitted in sandbox")
        return await self.sandbox.run_tool(tool, args)
```

#### Network Segmentation
```python
class NetworkPolicy:
    def __init__(self):
        self.policies: dict[str, NetworkPolicyRule] = {}
    
    def allow(self, from_agent: str, to: str, port: int = None, protocol: str = "tcp"):
        self.policies[f"{from_agent}->{to}:{port}"] = NetworkPolicyRule(
            from_agent=from_agent,
            to=to,
            port=port,
            protocol=protocol,
            action="allow"
        )
    
    def deny(self, from_agent: str, to: str, port: int = None):
        self.policies[f"{from_agent}->{to}:{port}"] = NetworkPolicyRule(
            from_agent=from_agent,
            to=to,
            action="deny"
        )
    
    def check(self, from_agent: str, to: str, port: int) -> bool:
        rule = self.policies.get(f"{from_agent}->{to}:{port}")
        if rule:
            return rule.action == "allow"
        return False  # Default deny
```

---

## Monitoring & Detection

### Anomaly Detection
```python
class AgentAnomalyDetector:
    def __init__(self, agents: dict[str, Agent]):
        self.agents = agents
        self.baselines: dict[str, BehaviorBaseline] = {}
    
    async def learn_baselines(self, observation_period: int = 3600):
        """Learn normal behavior patterns"""
        for agent_id, agent in self.agents.items():
            metrics = await self.collect_metrics(agent, observation_period)
            self.baselines[agent_id] = BehaviorBaseline(
                avg_tokens_per_min=statistics.mean(metrics.tokens_per_min),
                avg_tool_calls_per_min=statistics.mean(metrics.tool_calls_per_min),
                avg_latency_ms=statistics.mean(metrics.latency_ms),
                typical_tools=set(metrics.tools_used),
                typical_domains=set(metrics.domains_accessed)
            )
    
    async def detect_anomalies(self, agent_id: str) -> list[Anomaly]:
        baseline = self.baselines.get(agent_id)
        if not baseline:
            return []
        
        current = await self.collect_metrics(self.agents[agent_id], 60)
        anomalies = []
        
        # Token rate anomaly
        if current.tokens_per_min > baseline.avg_tokens_per_min * 3:
            anomalies.append(Anomaly(
                type="token_spike",
                severity="high",
                message=f"Token rate {current.tokens_per_min:.1f} > 3x baseline"
            ))
        
        # Unusual tool usage
        new_tools = set(current.tools_used) - baseline.typical_tools
        if new_tools:
            anomalies.append(Anomaly(
                type="new_tool_usage",
                severity="medium",
                message=f"New tools used: {new_tools}"
            ))
        
        return anomalies
```

### Cascading Failure Detection
```python
class CascadingFailureDetector:
    def __init__(self, agents: dict[str, Agent]):
        self.agents = agents
        self.failure_graph: nx.DiGraph = nx.DiGraph()
    
    def record_failure(self, agent_id: str, error: Exception, 
                       triggered_by: str = None):
        self.failure_graph.add_node(agent_id)
        if triggered_by:
            self.failure_graph.add_edge(triggered_by, agent_id)
    
    def detect_cascade(self) -> CascadeInfo:
        # Find cascading paths
        cascades = []
        for node in self.failure_graph.nodes():
            if self.failure_graph.in_degree(node) > 2:
                # Find path
                paths = list(nx.all_simple_paths(self.failure_graph, 
                                                  source=min(self.failure_graph.predecessors(node)),
                                                  target=node))
                if paths:
                    cascades.append(CascadeInfo(
                        target=node,
                        path=paths[0],
                        length=len(paths[0])
                    ))
        return CascadeInfo(cascades=cascades)
```

---

## Failure Modes & Mitigation

### Common Failure Modes

| Failure Mode | Description | Detection | Mitigation |
|---|---|---|---|
| **Cascading Failure** | One agent failure triggers others | Failure graph analysis | Circuit breakers, isolation |
| **Resource Exhaustion** | Agent consumes all resources | Resource monitoring | Quotas, limits |
| **Infinite Loop** | Agents trigger each other | Loop detection | Max depth, visited tracking |
| **Deadlock** | Circular wait | Wait-for graph | Timeouts, ordering |
| **Livelock** | Continuous state changes | Progress monitoring | Randomization, backoff |
| **Byzantine Behavior** | Malicious/buggy agent | Consensus verification | BFT consensus |
| **Data Corruption** | Shared state corrupted | Checksums, versioning | Immutable state, CRDTs |

### Deadlock Prevention
```python
class DeadlockPrevention:
    def __init__(self, agents: dict[str, Agent]):
        self.agents = agents
        self.locks: dict[str, asyncio.Lock] = {}
        self.lock_order: list[str] = []  # Global lock ordering
    
    def acquire_locks(self, agent_id: str, resources: list[str]):
        # Always acquire in global order
        ordered = sorted(resources, key=lambda r: self.lock_order.index(r))
        for resource in ordered:
            self.locks[resource].acquire()
    
    def release_locks(self, resources: list[str]):
        for resource in resources:
            self.locks[resource].release()
    
    def detect_deadlock(self) -> list[DeadlockCycle]:
        # Wait-for graph analysis
        wait_for = self.build_wait_for_graph()
        cycles = list(nx.simple_cycles(wait_for))
        return [DeadlockCycle(cycle=c) for c in cycles]
```

### Livelock Prevention
```python
class LivelockPrevention:
    def __init__(self):
        self.state_history: dict[str, list[AgentState]] = defaultdict(list)
        self.max_history = 100
    
    def record_state(self, agent_id: str, state: AgentState):
        self.state_history[agent_id].append(state)
        if len(self.state_history[agent_id]) > self.max_history:
            self.state_history[agent_id].pop(0)
    
    def detect_livelock(self, agent_id: str) -> bool:
        history = self.state_history[agent_id]
        if len(history) < 10:
            return False
        
        # Check for repeating pattern
        recent = history[-10:]
        for i in range(len(recent) - 1):
            if self._states_similar(recent[i], recent[i+1]):
                # Check if pattern repeats
                pattern = [self._state_signature(s) for s in recent]
                if self._has_repeating_pattern(pattern):
                    return True
        return False
    
    def break_livelock(self, agent_id: str) -> BreakAction:
        # Add randomness, backoff, or force state change
        return BreakAction(
            type="randomize",
            parameters={"seed": random.random()}
        )
```

---

## Safety Testing

### Chaos Engineering
```python
class AgentChaosEngineering:
    def __init__(self, system: MultiAgentSystem):
        self.system = system
    
    async def inject_failure(self, experiment: ChaosExperiment) -> ExperimentResult:
        # 1. Baseline
        baseline = await self.measure_steady_state()
        
        # 2. Inject
        if experiment.type == "agent_crash":
            await self.crash_agent(experiment.target_agent)
        elif experiment.type == "network_partition":
            await self.partition_network(experiment.target_agents)
        elif experiment.type == "resource_exhaustion":
            await self.exhaust_resources(experiment.target_agent)
        elif experiment.type == "latency_injection":
            await self.inject_latency(experiment.target_agent, experiment.latency_ms)
        
        # 3. Observe
        await asyncio.sleep(experiment.duration)
        impacted = await self.measure_impact()
        
        # 3. Recover
        await self.recover(experiment)
        
        return ExperimentResult(
            baseline=baseline,
            impacted=impacted,
            recovered=await self.measure_recovery()
        )
```

### Safety Test Suite
```python
class SafetyTestSuite:
    def __init__(self, system: MultiAgentSystem):
        self.system = system
    
    async def run_all_tests(self) -> SafetyTestReport:
        results = []
        
        # Test 1: Single agent failure containment
        results.append(await self.test_failure_containment())
        
        # Test 2: Cascading failure prevention
        results.append(await self.test_cascading_failure_prevention())
        
        # Test 3: Resource quota enforcement
        results.append(await self.test_resource_quotas())
        
        # Test 4: Circuit breaker activation
        results.append(await self.test_circuit_breakers())
        
        # Test 5: Deadlock detection
        results.append(await self.test_deadlock_detection())
        
        # Test 6: Livelock detection
        results.append(await self.test_livelock_detection())
        
        # Test 7: Byzantine agent isolation
        results.append(await self.test_byzantine_isolation())
        
        return SafetyTestReport(results=results)
```

---

## Safety Checklist

### Pre-Deployment
- [ ] Individual agent guardrails configured
- [ ] Inter-agent authorization policies defined
- [ ] Resource quotas set per agent
- [ ] Circuit breakers configured
- [ ] Network policies defined
- [ ] Sandbox environments configured
- [ ] Anomaly detection baselines established
- [ ] Failure injection tests passed
- [ ] Rollback procedures documented
- [ ] Incident response plan ready

### Runtime Monitoring
- [ ] Real-time anomaly detection active
- [ ] Cascading failure detection active
- [ ] Resource utilization alerts configured
- [ ] Circuit breaker dashboards visible
- [ ] Audit logging enabled
- [ ] Human escalation paths defined

### Incident Response
- [ ] Kill switch for each agent
- [ ] System-wide emergency stop
- [ ] State rollback procedures
- [ ] Communication plan for stakeholders
- [ ] Post-incident review process

---

## Related Notes

- [[MultiAgent/Coordination-Patterns.md]] — Safe coordination patterns
- [[MultiAgent/Communication-Protocols.md]] — Secure communication
- [[MultiAgent/Delegation-Handoff.md]] — Safe handoffs
- [[MultiAgent/Negotiation-Consensus.md]] — Safe negotiation
- [[MultiAgent/Emergent-Behavior.md]] — Controlling emergence
- [[Production/Reliability.md]] — Circuit breakers, reliability
- [[Production/Security.md]] — Security practices
- [[Production/Observability.md]] — Monitoring
- [[Production/Reliability.md]] — Reliability patterns
- [[Security/Secure-Coding.md]] — Secure coding
- [[Security/Supply-Chain-Security.md]] — Supply chain safety
- [[Frontiers/Agent-Societies.md]] — Societal safety
- [[Frontiers/Research-Directions.md]] — Safety research

---

*Sources: "System Safety" (Leveson), "Designing Distributed Systems" (Burns), "Release It!" (Nygard), "Site Reliability Engineering" (Google), Chaos Engineering (Netflix), Distributed Systems Safety, Circuit Breaker Pattern, Byzantine Fault Tolerance, Deadlock/Livelock prevention, Chaos Engineering principles.*