Safety & Coordination

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

Safety Challenges in Multi-Agent Systems

Safety Layers

1. Individual Agent Guardrails

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

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

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

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

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

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

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

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

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

Deadlock Prevention

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

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

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

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

• Coordination Patterns — Safe coordination patterns
• Communication Protocols — Secure communication
• Delegation Handoff — Safe handoffs
• Negotiation Consensus — Safe negotiation
• Emergent Behavior — Controlling emergence
• Reliability — Circuit breakers, reliability
• Security — Security practices
• Observability — Monitoring
• Reliability — Reliability patterns
• Secure Coding — Secure coding
• Supply Chain Security — Supply chain safety
• Agent Societies — Societal safety
• Research Directions — 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.