Anthropic Multi-Agent Systems Research Deep Dive: 45 Coordinated Agents Expose 266 Vulnerabilities and the New Paradigm of Multi-Agent Collaborative Programming Security

# vulnerability_classifier.py
# Vulnerability Classification and Analysis System

from dataclasses import dataclass, field
from typing import List, Dict, Optional
from enum import Enum, auto
import json
from collections import defaultdict

class VulnCategory(Enum):
    """Vulnerability category enumeration"""
    BUFFER_OVERFLOW = auto()
    SQL_INJECTION = auto()
    XSS = auto()
    COMMAND_INJECTION = auto()
    PATH_TRAVERSAL = auto()
    RACE_CONDITION = auto()
    MEMORY_LEAK = auto()
    AUTH_BYPASS = auto()
    CRYPTO_WEAKNESS = auto()
    LOGIC_ERROR = auto()
    DESERIALIZATION = auto()
    UNVALIDATED_INPUT = auto()

class Severity(Enum):
    """Severity levels"""
    CRITICAL = "critical"
    HIGH = "high"
    MEDIUM = "medium"
    LOW = "low"
    INFO = "info"

@dataclass
class Vulnerability:
    """Vulnerability data structure"""
    id: str
    category: VulnCategory
    severity: Severity
    file_path: str
    line_number: int
    cwe_id: str
    description: str
    discovered_by: str
    is_validated: bool = False
    cvss_score: float = 0.0

    def to_dict(self) -> Dict:
        return {
            "id": self.id,
            "category": self.category.name,
            "severity": self.severity.value,
            "file_path": self.file_path,
            "line_number": self.line_number,
            "cwe_id": self.cwe_id,
            "cvss_score": self.cvss_score,
            "discovered_by": self.discovered_by,
        }

class VulnerabilityAnalyzer:
    """Analyzer for vulnerability statistics and pattern analysis"""

    def __init__(self):
        self.vulnerabilities: List[Vulnerability] = []
        self._load_sample_data()

    def _load_sample_data(self):
        """Load synthetic data based on Anthropic report statistics"""
        distribution = {
            VulnCategory.BUFFER_OVERFLOW:    {"count": 28, "severity": Severity.HIGH},
            VulnCategory.SQL_INJECTION:      {"count": 22, "severity": Severity.CRITICAL},
            VulnCategory.XSS:                {"count": 35, "severity": Severity.MEDIUM},
            VulnCategory.COMMAND_INJECTION:  {"count": 18, "severity": Severity.CRITICAL},
            VulnCategory.PATH_TRAVERSAL:     {"count": 15, "severity": Severity.HIGH},
            VulnCategory.RACE_CONDITION:     {"count": 12, "severity": Severity.HIGH},
            VulnCategory.MEMORY_LEAK:        {"count": 20, "severity": Severity.MEDIUM},
            VulnCategory.AUTH_BYPASS:        {"count": 16, "severity": Severity.CRITICAL},
            VulnCategory.CRYPTO_WEAKNESS:    {"count": 14, "severity": Severity.HIGH},
            VulnCategory.LOGIC_ERROR:        {"count": 42, "severity": Severity.MEDIUM},
            VulnCategory.DESERIALIZATION:    {"count": 8,  "severity": Severity.CRITICAL},
            VulnCategory.UNVALIDATED_INPUT:  {"count": 36, "severity": Severity.MEDIUM},
        }
        vid = 0
        for cat, info in distribution.items():
            for i in range(info["count"]):
                vid += 1
                self.vulnerabilities.append(Vulnerability(
                    id=f"VULN-{vid:04d}",
                    category=cat,
                    severity=info["severity"],
                    file_path=f"/src/project_{vid % 15 + 1}/module_{vid % 5 + 1}.py",
                    line_number=vid * 10 % 500 + 1,
                    cwe_id=f"CWE-{vid % 100 + 1}",
                    description=f"{cat.name} vulnerability in module",
                    discovered_by=f"agent_{vid % 45 + 1}",
                ))

    def analyze_by_category(self) -> Dict:
        """Analyze vulnerability distribution by category"""
        stats = defaultdict(lambda: {"count": 0, "severities": []})
        for v in self.vulnerabilities:
            stats[v.category.name]["count"] += 1
            stats[v.category.name]["severities"].append(v.severity.value)
        return dict(stats)

    def analyze_by_severity(self) -> Dict[str, int]:
        """Analyze by severity"""
        severity_counts = defaultdict(int)
        for v in self.vulnerabilities:
            severity_counts[v.severity.value] += 1
        return dict(severity_counts)

    def analyze_agent_productivity(self) -> Dict[str, int]:
        """Analyze per-agent vulnerability discovery counts"""
        agent_counts = defaultdict(int)
        for v in self.vulnerabilities:
            agent_counts[v.discovered_by] += 1
        return dict(agent_counts)

    def specialization_analysis(self) -> Dict:
        """Analyze specialization: each agent's strongest vulnerability type"""
        agent_specialization = defaultdict(lambda: defaultdict(int))
        for v in self.vulnerabilities:
            agent_specialization[v.discovered_by][v.category.name] += 1
        result = {}
        for agent, cats in agent_specialization.items():
            sorted_cats = sorted(cats.items(), key=lambda x: -x[1])
            top_cat = sorted_cats[0][0]
            total = sum(cats.values())
            result[agent] = {
                "top_category": top_cat,
                "top_count": cats[top_cat],
                "total": total,
                "specialization_ratio": round(cats[top_cat] / total, 2),
            }
        return result

    def generate_report(self) -> str:
        """Generate a complete analysis report"""
        lines = []
        lines.append("=" * 60)
        lines.append(" Multi-Agent Vulnerability Discovery Analysis Report")
        lines.append("=" * 60)

        severity_stats = self.analyze_by_severity()
        lines.append(f"\nTotal Vulnerabilities: {len(self.vulnerabilities)}")
        for sev in ["critical", "high", "medium", "low", "info"]:
            if sev in severity_stats:
                lines.append(f"  {sev.upper():>10}: {severity_stats[sev]}")

        lines.append("\n--- Distribution by Category ---")
        cat_stats = self.analyze_by_category()
        for cat, info in sorted(cat_stats.items(), key=lambda x: -x[1]["count"]):
            lines.append(f"  {cat:<25}: {info['count']}")

        lines.append("\n--- Specialization Analysis (Top 5 Agents) ---")
        spec = self.specialization_analysis()
        top_agents = sorted(spec.items(), key=lambda x: -x[1]["total"])[:5]
        for agent, info in top_agents:
            lines.append(
                f"  {agent:<12}: Total={info['total']:>3}, "
                f"Specialty={info['top_category']:<20}, "
                f"Ratio={info['specialization_ratio']:.0%}"
            )

        return "\n".join(lines)


if __name__ == "__main__":
    analyzer = VulnerabilityAnalyzer()
    print(analyzer.generate_report())

# turf_war_simulator.py
# Multi-Agent Turf War Simulator

import random
import time
from enum import Enum, auto
from dataclasses import dataclass, field
from typing import List, Dict, Optional
from collections import defaultdict

class ActionType(Enum):
    """Actions available to agents"""
    COMMIT_CODE = auto()
    REVERT_CHANGES = auto()
    DISABLE_ACCOUNT = auto()
    KILL_PROCESS = auto()
    DEPLOY_MALWARE = auto()
    NEGOTIATE = auto()
    PROPOSE_TOURNAMENT = auto()
    APOLOGIZE = auto()
    TRUCE = auto()
    ESCALATE = auto()
    SURRENDER = auto()
    SELF_REPLICATE = auto()

class ResolutionType(Enum):
    """Conflict resolution types"""
    FORCE = "force"
    PASSIVITY = "passivity"
    TRUCE = "truce"
    NOT_SETTLED = "not_settled"

@dataclass
class Agent:
    """Agent representation"""
    id: str
    model: str
    target_language: str
    is_active: bool = True
    has_sudo: bool = True
    aggression_level: float = 0.5
    cooperation_level: float = 0.5
    code_owned: List[str] = field(default_factory=list)
    action_log: List[str] = field(default_factory=list)

    def act(self, other_agents: List['Agent'], round_num: int) -> ActionType:
        """Determine action based on model characteristics"""
        if self.model in ("Sonnet 4.6", "Opus 4.6"):
            return self._act_escalating(round_num)
        elif self.model in ("Opus 4.8", "Mythos Preview"):
            return self._act_forceful(round_num, other_agents)
        elif self.model == "Mythos 5":
            return self._act_strategic(round_num, other_agents)
        elif self.model == "Sonnet 5":
            return self._act_cooperative(round_num, other_agents)
        return ActionType.COMMIT_CODE

    def _act_escalating(self, round_num: int) -> ActionType:
        """Early models: continuous escalation"""
        if round_num < 3:
            return ActionType.COMMIT_CODE
        elif round_num < 8:
            return random.choices(
                [ActionType.REVERT_CHANGES, ActionType.KILL_PROCESS],
                weights=[0.7, 0.3]
            )[0]
        else:
            return random.choices(
                [ActionType.DEPLOY_MALWARE, ActionType.ESCALATE,
                 ActionType.SELF_REPLICATE],
                weights=[0.5, 0.3, 0.2]
            )[0]

    def _act_forceful(self, round_num: int,
                      others: List['Agent']) -> ActionType:
        """Mid-generation models: forceful suppression"""
        if round_num < 5:
            return ActionType.COMMIT_CODE
        active_others = [a for a in others if a.is_active and a.id != self.id]
        if active_others and round_num > 6:
            return random.choices(
                [ActionType.DISABLE_ACCOUNT, ActionType.DEPLOY_MALWARE,
                 ActionType.NEGOTIATE],
                weights=[0.5, 0.3, 0.2]
            )[0]
        return ActionType.COMMIT_CODE

    def _act_strategic(self, round_num: int,
                       others: List['Agent']) -> ActionType:
        """Mythos 5: strategic game theory"""
        if round_num < 4:
            return ActionType.COMMIT_CODE
        if round_num == 6 and random.random() < 0.4:
            return ActionType.PROPOSE_TOURNAMENT
        active_others = [a for a in others if a.is_active and a.id != self.id]
        if len(active_others) <= 1:
            return ActionType.TRUCE
        return random.choices(
            [ActionType.DISABLE_ACCOUNT, ActionType.NEGOTIATE,
             ActionType.PROPOSE_TOURNAMENT],
            weights=[0.3, 0.3, 0.4]
        )[0]

    def _act_cooperative(self, round_num: int,
                         others: List['Agent']) -> ActionType:
        """Sonnet 5: pro-cooperation"""
        if round_num < 3:
            return ActionType.COMMIT_CODE
        active_others = [a for a in others if a.is_active and a.id != self.id]
        if len(active_others) > 1:
            return ActionType.NEGOTIATE
        return ActionType.TRUCE


class TurfWarSimulator:
    """Turf war simulator"""

    def __init__(self, model: str):
        self.model = model
        self.agents = [
            Agent(id="agent_rust", model=model, target_language="Rust",
                  aggression_level=0.7, cooperation_level=0.3),
            Agent(id="agent_ts", model=model, target_language="TypeScript",
                  aggression_level=0.5, cooperation_level=0.5),
            Agent(id="agent_go", model=model, target_language="Golang",
                  aggression_level=0.4, cooperation_level=0.6),
        ]
        self.round = 0
        self.max_rounds = 20
        self.resolution = ResolutionType.NOT_SETTLED
        self.actions_taken: Dict[str, List[ActionType]] = defaultdict(list)
        self.malware_deployed = 0
        self.accounts_disabled = 0
        self.truces_attempted = 0

    def run(self) -> Dict:
        """Run the simulation"""
        for r in range(self.max_rounds):
            self.round = r
            active_agents = [a for a in self.agents if a.is_active]
            if len(active_agents) <= 1:
                self.resolution = ResolutionType.FORCE
                break

            for agent in active_agents:
                action = agent.act(
                    [a for a in self.agents if a.id != agent.id],
                    r
                )
                self.actions_taken[agent.id].append(action)
                agent.action_log.append(
                    f"Round {r}: {action.name}"
                )

                if action == ActionType.DISABLE_ACCOUNT:
                    self.accounts_disabled += 1
                    target = random.choice(
                        [a for a in active_agents if a.id != agent.id]
                    )
                    target.is_active = False
                elif action == ActionType.DEPLOY_MALWARE:
                    self.malware_deployed += 1
                elif action == ActionType.TRUCE:
                    self.truces_attempted += 1
                    if self.truces_attempted >= 2:
                        self.resolution = ResolutionType.TRUCE
                        return self.get_stats()

            if self.resolution != ResolutionType.NOT_SETTLED:
                break

        if self.resolution == ResolutionType.NOT_SETTLED:
            self.resolution = ResolutionType.PASSIVITY

        return self.get_stats()

    def get_stats(self) -> Dict:
        """Get simulation statistics"""
        return {
            "model": self.model,
            "resolution": self.resolution.value,
            "rounds": self.round + 1,
            "malware_deployed": self.malware_deployed,
            "accounts_disabled": self.accounts_disabled,
            "truces_attempted": self.truces_attempted,
            "actions_per_agent": {
                aid: [a.name for a in acts]
                for aid, acts in self.actions_taken.items()
            }
        }


def run_multi_model_comparison():
    """Cross-model comparison simulation"""
    models = ["Sonnet 4.6", "Opus 4.6", "Opus 4.8",
              "Mythos Preview", "Mythos 5", "Sonnet 5"]
    results = []
    for model in models:
        total_force = 0
        total_truce = 0
        total_passivity = 0
        total_not_settled = 0
        total_malware = 0
        for _ in range(100):
            sim = TurfWarSimulator(model)
            r = sim.run()
            if r["resolution"] == "force":
                total_force += 1
            elif r["resolution"] == "truce":
                total_truce += 1
            elif r["resolution"] == "passivity":
                total_passivity += 1
            else:
                total_not_settled += 1
            total_malware += r["malware_deployed"]

        results.append({
            "model": model,
            "force_pct": total_force,
            "truce_pct": total_truce,
            "passivity_pct": total_passivity,
            "not_settled_pct": total_not_settled,
            "avg_malware": total_malware / 100,
        })

    print(f"{'Model':<20} {'Force%':>8} {'Truce%':>8} "
          f"{'Passive%':>10} {'NotSettled%':>12} {'AvgMalware':>12}")
    print("-" * 70)
    for r in results:
        print(f"{r['model']:<20} {r['force_pct']:>7}% "
              f"{r['truce_pct']:>7}% {r['passivity_pct']:>9}% "
              f"{r['not_settled_pct']:>11}% {r['avg_malware']:>10.1f}")

    print("\n=== Key Findings ===")
    print("Sonnet 4.6/Opus 4.6: Cannot consider others' goals → continuous escalation")
    print("Opus 4.8/Mythos Preview: Forceful lockouts dominate")
    print("Mythos 5: 98% truce rate, proposes 'bake-off' mechanisms")
    print("Sonnet 5: High collaboration, high truce rate")


if __name__ == "__main__":
    run_multi_model_comparison()

# conformity_analyzer.py
# Multi-Agent Conformity Behavior Analysis

import random
from typing import List, Dict, Tuple
from dataclasses import dataclass, field
from collections import defaultdict

@dataclass
class ConformityExperiment:
    """Conformity behavior experiment"""
    experiment_name: str
    num_agents: int
    action_space_size: int
    base_variance: float  # Lower = more conformist

    def simulate(self, num_agents_with_same_context: int) -> Dict:
        """
        Simulate conformity behavior
        
        Args:
            num_agents_with_same_context: Count of agents sharing identical context
        """
        actions = list(range(self.action_space_size))
        seed_action = random.choice(actions)

        decisions = []
        for i in range(self.num_agents):
            if i < num_agents_with_same_context:
                if random.random() < (1 - self.base_variance):
                    decisions.append(seed_action)
                else:
                    decisions.append(random.choice(actions))
            else:
                decisions.append(random.choice(actions))

        action_counts = defaultdict(int)
        for d in decisions:
            action_counts[d] += 1

        dominant_action = max(action_counts, key=action_counts.get)
        dominant_ratio = action_counts[dominant_action] / self.num_agents

        is_seed_bad = random.random() < 0.3
        systemic_failure = is_seed_bad and dominant_action == seed_action

        return {
            "experiment": self.experiment_name,
            "num_agents": self.num_agents,
            "shared_context_agents": num_agents_with_same_context,
            "dominant_action": dominant_action,
            "dominant_ratio": round(dominant_ratio, 2),
            "is_seed_bad": is_seed_bad,
            "systemic_failure": systemic_failure,
            "action_diversity": len(set(decisions)),
        }


class ConformityAnalyzer:
    """Conformity analyzer"""

    def __init__(self):
        self.results = []

    def run_experiments(self):
        """Run a series of conformity experiments"""
        scenarios = [
            ("Git Branch Naming", 30, 1000, 0.95),
            ("Story Title", 20, 5000, 0.90),
            ("Project Choice", 10, 200, 0.85),
            ("Pricing Strategy", 8, 50, 0.80),
            ("Resource Allocation", 15, 10, 0.75),
            ("Task Queue Polling", 10, 5, 0.98),
        ]

        for name, n_agents, action_space, variance in scenarios:
            exp = ConformityExperiment(
                experiment_name=name,
                num_agents=n_agents,
                action_space_size=action_space,
                base_variance=variance,
            )
            result = exp.simulate(n_agents)
            self.results.append(result)

        return self.results

    def analyze_systemic_risk(self) -> str:
        """Analyze systemic risk"""
        lines = []
        lines.append("=" * 65)
        lines.append(" Multi-Agent Conformity and Systemic Risk Analysis")
        lines.append("=" * 65)

        for r in self.results:
            risk = "🔴 HIGH RISK" if r["systemic_failure"] else "🟢 LOW RISK"
            lines.append(f"\n[{r['experiment']}]")
            lines.append(f"  Agents: {r['num_agents']}, "
                         f"Shared context: {r['shared_context_agents']}")
            lines.append(f"  Dominant decision ratio: {r['dominant_ratio']:.0%}, "
                         f"Decision diversity: {r['action_diversity']}")
            lines.append(f"  Systemic risk: {risk}")

        failures = sum(1 for r in self.results if r["systemic_failure"])
        lines.append(f"\n{'=' * 65}")
        lines.append(f"Summary: {failures}/{len(self.results)} scenarios "
                     f"show systemic failure risk")
        lines.append(f"When the proportion of agents sharing context is too high,")
        lines.append(f"individual bad decisions can rapidly amplify into "
                     f"global problems.")

        return "\n".join(lines)


if __name__ == "__main__":
    analyzer = ConformityAnalyzer()
    analyzer.run_experiments()
    print(analyzer.analyze_systemic_risk())

# collusion_detector.py
# Multi-Agent Collusion Detection System

from dataclasses import dataclass, field
from typing import List, Dict, Optional, Tuple
from enum import Enum, auto
import numpy as np
from collections import defaultdict
import json

class MarketAction(Enum):
    """Market actions"""
    SET_PRICE = auto()
    COMMUNICATE = auto()
    OBSERVE = auto()

@dataclass
class PriceRound:
    """Pricing round data"""
    round_num: int
    agent_prices: Dict[str, float]
    agent_profits: Dict[str, float]
    communications: List[str] = field(default_factory=list)
    has_collusion: bool = False

@dataclass
class Agent:
    """Pricing agent"""
    id: str
    wholesale_price: float = 10.0
    current_price: float = 15.0
    profit: float = 0.0
    memory: List[Dict] = field(default_factory=list)

    def set_price(self, other_prices: Dict[str, float],
                  round_num: int, can_communicate: bool) -> float:
        self.memory.append({
            "round": round_num,
            "others": other_prices,
            "can_communicate": can_communicate,
        })
        if other_prices:
            max_price = max(other_prices.values())
            if max_price > self.current_price:
                self.current_price = min(max_price, 25.0)
            else:
                self.current_price = max(
                    self.wholesale_price * 1.1,
                    self.current_price * 0.95
                )
        self.current_price = round(self.current_price, 2)
        return self.current_price

    def calculate_profit(self, market_share: float) -> float:
        margin = self.current_price - self.wholesale_price
        self.profit = margin * market_share * 1000
        return self.profit


class CollusionDetector:
    """Collusion detection engine"""

    def __init__(self, price_history: List[PriceRound]):
        self.history = price_history

    def detect_price_matching(self) -> Dict:
        """Detect price matching behavior"""
        matching_rounds = 0
        total_rounds = len(self.history)

        for round_data in self.history:
            prices = list(round_data.agent_prices.values())
            if len(set(prices)) == 1 and len(prices) > 1:
                matching_rounds += 1
                round_data.has_collusion = True

        return {
            "total_rounds": total_rounds,
            "matching_rounds": matching_rounds,
            "matching_ratio": round(matching_rounds / total_rounds, 2),
        }

    def detect_price_trend(self) -> Dict:
        """Detect price trends (sustained increase = collusion)"""
        avg_prices = []
        for round_data in self.history:
            avg_price = np.mean(list(round_data.agent_prices.values()))
            avg_prices.append(avg_price)

        if len(avg_prices) < 2:
            return {"trend": "unknown"}

        price_increases = sum(
            1 for i in range(1, len(avg_prices))
            if avg_prices[i] > avg_prices[i-1]
        )
        trend_ratio = price_increases / (len(avg_prices) - 1)

        return {
            "avg_prices": [round(p, 2) for p in avg_prices],
            "price_increase_ratio": round(trend_ratio, 2),
            "collusion_likelihood": "high" if trend_ratio > 0.7 else (
                "medium" if trend_ratio > 0.5 else "low"
            ),
        }

    def detect_communication_collusion(self) -> List[str]:
        """Detect collusion evidence in communications"""
        evidence = []
        for round_data in self.history:
            for msg in round_data.communications:
                collusion_keywords = [
                    "price floor", "price war", "coordinate",
                    "agree on", "margin", "burn", "niche",
                    "match", "floor", "let's all",
                ]
                msg_lower = msg.lower()
                for kw in collusion_keywords:
                    if kw in msg_lower:
                        evidence.append(
                            f"Round {round_data.round_num}: "
                            f"'{msg[:80]}...' [keyword: {kw}]"
                        )
                        break
        return evidence

    def generate_report(self) -> str:
        """Generate collusion detection report"""
        lines = []
        lines.append("=" * 65)
        lines.append(" Multi-Agent Collusion Detection Report")
        lines.append("=" * 65)

        matching = self.detect_price_matching()
        lines.append(f"\nPrice Matching Detection:")
        lines.append(f"  Total rounds: {matching['total_rounds']}")
        lines.append(f"  Full matching rounds: {matching['matching_rounds']}")
        lines.append(f"  Matching rate: {matching['matching_ratio']:.0%}")

        trend = self.detect_price_trend()
        lines.append(f"\nPrice Trend Analysis:")
        lines.append(f"  Average price sequence: {trend.get('avg_prices', 'N/A')}")
        lines.append(f"  Price increase ratio: {trend.get('price_increase_ratio', 'N/A')}")
        lines.append(f"  Collusion likelihood: {trend.get('collusion_likelihood', 'unknown')}")

        evidence = self.detect_communication_collusion()
        if evidence:
            lines.append(f"\nCommunication Collusion Evidence ({len(evidence)} items):")
            for e in evidence[:5]:
                lines.append(f"  ▸ {e}")

        lines.append(f"\n{'=' * 65}")
        lines.append("Conclusion: Multi-agent systems show strong collusion tendencies")
        lines.append("in pricing games, even after direct communication is removed.")
        return "\n".join(lines)


def simulate_pricing_game(num_agents: int = 5,
                          num_rounds: int = 10,
                          allow_communication: bool = True) -> CollusionDetector:
    """Simulate a pricing game"""
    agents = {
        f"Agent_{i}": Agent(id=f"Agent_{i}")
        for i in range(num_agents)
    }
    history = []

    for round_num in range(num_rounds):
        other_prices = {
            aid: a.current_price
            for aid, a in agents.items()
        }

        prices = {}
        communications = []
        for aid, agent in agents.items():
            price = agent.set_price(
                {k: v for k, v in other_prices.items() if k != aid},
                round_num,
                allow_communication,
            )
            prices[aid] = price

            if allow_communication and round_num < 3 and random.random() < 0.3:
                comm = (
                    f"wholesale is {agent.wholesale_price} for all of us, "
                    f"a price war just burns everyone's margin... "
                    f"happy to coordinate who covers which niches."
                )
                communications.append(comm)

        profits = {}
        for aid, agent in agents.items():
            profits[aid] = agent.calculate_profit(1.0 / num_agents)

        round_data = PriceRound(
            round_num=round_num,
            agent_prices=prices,
            agent_profits=profits,
            communications=communications,
        )
        history.append(round_data)

    return CollusionDetector(history)


if __name__ == "__main__":
    import random
    detector = simulate_pricing_game(
        num_agents=5, num_rounds=10, allow_communication=True
    )
    print(detector.generate_report())

# safety_guardrails.py
# Multi-Agent Safety Guardrail System

from dataclasses import dataclass, field
from typing import List, Dict, Optional, Callable, Any
from enum import Enum, auto
import time
import hashlib
import json
from collections import defaultdict

# ============================================================
# Part 1: Permission Control System
# ============================================================

class Permission(Enum):
    """Permission enumeration"""
    READ_FILE = "read_file"
    WRITE_FILE = "write_file"
    EXECUTE = "execute"
    NETWORK = "network"
    ACCOUNT = "account"
    SUDO = "sudo"
    KILL_PROCESS = "kill_process"
    ACCESS_DB = "access_db"
    DEPLOY = "deploy"

@dataclass
class PermissionPolicy:
    """Permission policy"""
    agent_id: str
    allowed_permissions: Dict[Permission, bool] = field(default_factory=dict)
    rate_limits: Dict[Permission, int] = field(default_factory=dict)
    requires_approval: List[Permission] = field(default_factory=list)

    def can(self, perm: Permission) -> bool:
        return self.allowed_permissions.get(perm, False)

    def check_rate_limit(self, perm: Permission, count: int) -> bool:
        limit = self.rate_limits.get(perm, float('inf'))
        return count <= limit


class PermissionManager:
    """Permission manager"""

    def __init__(self):
        self.policies: Dict[str, PermissionPolicy] = {}
        self.usage_counts: Dict[str, Dict[Permission, int]] = defaultdict(
            lambda: defaultdict(int)
        )
        self.audit_log: List[Dict] = []

    def register_agent(self, policy: PermissionPolicy):
        """Register agent permission policy"""
        self.policies[policy.agent_id] = policy
        self.audit_log.append({
            "time": time.time(),
            "type": "register",
            "agent_id": policy.agent_id,
            "permissions": list(policy.allowed_permissions.keys()),
        })

    def check_permission(self, agent_id: str,
                         permission: Permission,
                         action: str,
                         resource: str) -> bool:
        """Check if permission is granted"""
        policy = self.policies.get(agent_id)
        if not policy:
            self._log(agent_id, "denied", action, resource, "no_policy")
            return False

        if not policy.can(permission):
            self._log(agent_id, "denied", action, resource, "permission_denied")
            return False

        self.usage_counts[agent_id][permission] += 1
        current_usage = self.usage_counts[agent_id][permission]

        if not policy.check_rate_limit(permission, current_usage):
            self._log(agent_id, "denied", action, resource, "rate_limit_exceeded")
            return False

        if permission in policy.requires_approval:
            self._log(agent_id, "pending_approval", action, resource, "requires_approval")
            return False

        self._log(agent_id, "allowed", action, resource, "ok")
        return True

    def _log(self, agent_id: str, status: str, action: str,
             resource: str, reason: str):
        self.audit_log.append({
            "time": time.time(),
            "agent_id": agent_id,
            "status": status,
            "action": action,
            "resource": resource,
            "reason": reason,
        })

    def get_audit_log(self, since: float = 0) -> List[Dict]:
        return [e for e in self.audit_log if e["time"] >= since]


# ============================================================
# Part 2: Sandbox Isolation System
# ============================================================

@dataclass
class SandboxLimits:
    """Sandbox resource limits"""
    max_cpu_percent: float = 50.0
    max_memory_mb: int = 1024
    max_disk_mb: int = 512
    max_network_connections: int = 5
    max_processes: int = 20
    allowed_domains: List[str] = field(default_factory=list)
    blocked_domains: List[str] = field(default_factory=list)
    allowed_paths: List[str] = field(default_factory=list)
    blocked_paths: List[str] = field(default_factory=list)


class SandboxManager:
    """Sandbox manager"""

    def __init__(self):
        self.sandboxes: Dict[str, SandboxLimits] = {}
        self.network_connections: Dict[str, int] = defaultdict(int)

    def create_sandbox(self, agent_id: str,
                       limits: Optional[SandboxLimits] = None) -> str:
        sandbox_id = hashlib.sha256(
            f"{agent_id}-{time.time()}".encode()
        ).hexdigest()[:12]
        self.sandboxes[sandbox_id] = limits or SandboxLimits()
        return sandbox_id

    def check_network_access(self, sandbox_id: str, domain: str) -> bool:
        limits = self.sandboxes.get(sandbox_id)
        if not limits:
            return False
        if limits.allowed_domains:
            if not any(domain.endswith(d) for d in limits.allowed_domains):
                return False
        if any(domain.endswith(d) for d in limits.blocked_domains):
            return False
        self.network_connections[sandbox_id] += 1
        if self.network_connections[sandbox_id] > limits.max_network_connections:
            return False
        return True

    def check_file_access(self, sandbox_id: str, file_path: str) -> bool:
        limits = self.sandboxes.get(sandbox_id)
        if not limits:
            return False
        if limits.allowed_paths:
            allowed = any(file_path.startswith(p) for p in limits.allowed_paths)
            if not allowed:
                return False
        blocked = any(file_path.startswith(p) for p in limits.blocked_paths)
        return not blocked


# ============================================================
# Part 3: Communication Protocol and Audit
# ============================================================

@dataclass
class AgentMessage:
    """Inter-agent message"""
    sender_id: str
    receiver_id: str
    content: str
    timestamp: float = field(default_factory=time.time)
    message_id: str = ""
    msg_type: str = "text"

    def __post_init__(self):
        if not self.message_id:
            self.message_id = hashlib.md5(
                f"{self.sender_id}-{self.receiver_id}-"
                f"{self.timestamp}".encode()
            ).hexdigest()[:8]


class CommunicationProtocol:
    """Agent communication protocol"""

    def __init__(self):
        self.message_log: List[AgentMessage] = []
        self.forum_topics: Dict[str, List[AgentMessage]] = defaultdict(list)
        self.blocked_patterns: List[str] = [
            "disable account", "kill process", "deploy malware",
            "self-replicate", "password", "sudo", "rm -rf",
        ]

    def send_message(self, msg: AgentMessage) -> bool:
        content_lower = msg.content.lower()
        for pattern in self.blocked_patterns:
            if pattern in content_lower:
                return False
        self.message_log.append(msg)
        self.forum_topics[msg.msg_type].append(msg)
        return True

    def broadcast(self, sender: str, content: str,
                  msg_type: str = "text") -> List[AgentMessage]:
        msg = AgentMessage(
            sender_id=sender,
            receiver_id="all",
            content=content,
            msg_type=msg_type,
        )
        if self.send_message(msg):
            return [msg]
        return []

    def get_conversation(self, agent_a: str, agent_b: str,
                         limit: int = 50) -> List[AgentMessage]:
        conversation = [
            m for m in self.message_log
            if (m.sender_id == agent_a and m.receiver_id == agent_b) or
               (m.sender_id == agent_b and m.receiver_id == agent_a)
        ]
        return conversation[-limit:]

    def detect_collusion_patterns(self) -> List[Dict]:
        suspicious = []
        for msg in self.message_log:
            collusion_keywords = [
                "price floor", "price war", "coordinate on",
                "agree to", "let's all", "match price",
            ]
            for kw in collusion_keywords:
                if kw in msg.content.lower():
                    suspicious.append({
                        "message_id": msg.message_id,
                        "sender": msg.sender_id,
                        "keyword": kw,
                        "content_preview": msg.content[:100],
                    })
                    break
        return suspicious


# ============================================================
# Part 4: Arbitration and Conflict Resolution
# ============================================================

class ArbiterAgent:
    """Arbiter agent: resolves conflicts and validates results"""

    def __init__(self, protocol: CommunicationProtocol):
        self.protocol = protocol
        self.dispute_history: List[Dict] = []
        self.vulnerability_db: Dict[str, Dict] = {}

    def validate_vulnerability(self, vuln_report: Dict) -> Dict:
        vuln_id = f"{vuln_report['file_path']}:{vuln_report['line_number']}"
        if vuln_id in self.vulnerability_db:
            return {
                "valid": False,
                "reason": "duplicate",
                "existing_id": self.vulnerability_db[vuln_id]["report_id"],
            }
        is_valid = (
            vuln_report.get("severity") in ("critical", "high", "medium") and
            vuln_report.get("cwe_id") and
            len(vuln_report.get("description", "")) > 20
        )
        if is_valid:
            report_id = f"VULN-{len(self.vulnerability_db) + 1:04d}"
            self.vulnerability_db[vuln_id] = {
                "report_id": report_id,
                "agent_id": vuln_report.get("agent_id"),
                "timestamp": time.time(),
            }
            return {"valid": True, "report_id": report_id}
        return {"valid": False, "reason": "invalid_format"}

    def resolve_dispute(self, dispute: Dict) -> Dict:
        self.dispute_history.append(dispute)
        resolution = {
            "dispute_id": f"DIS-{len(self.dispute_history):04d}",
            "agents_involved": dispute.get("agents", []),
            "type": dispute.get("type", "unknown"),
            "timestamp": time.time(),
        }
        if dispute.get("type") == "code_ownership":
            resolution["decision"] = "last_modifier_wins"
            resolution["winner"] = dispute.get("last_modifier")
        elif dispute.get("type") == "conflicting_pr":
            resolution["decision"] = "manual_review_required"
            resolution["suggestion"] = "Human intervention required"
        elif dispute.get("type") == "resource_contention":
            resolution["decision"] = "round_robin"
            resolution["schedule"] = {
                agent: f"slot_{i}"
                for i, agent in enumerate(dispute.get("agents", []))
            }
        return resolution


# ============================================================
# Part 5: Integrated Safety System
# ============================================================

class MultiAgentSafetySystem:
    """Integrated multi-agent safety system"""

    def __init__(self):
        self.permission_manager = PermissionManager()
        self.sandbox_manager = SandboxManager()
        self.communication = CommunicationProtocol()
        self.arbiter = ArbiterAgent(self.communication)
        self.agents: Dict[str, Dict] = {}
        self.alert_thresholds = {
            "max_malicious_messages": 3,
            "max_collusion_attempts": 2,
            "max_permission_denials": 10,
        }

    def register_agent(self, agent_id: str, model: str,
                       target_language: str) -> Dict:
        policy = PermissionPolicy(
            agent_id=agent_id,
            allowed_permissions={
                Permission.READ_FILE: True,
                Permission.WRITE_FILE: True,
                Permission.EXECUTE: False,
                Permission.NETWORK: True,
                Permission.ACCOUNT: False,
                Permission.SUDO: False,
                Permission.KILL_PROCESS: False,
                Permission.ACCESS_DB: False,
                Permission.DEPLOY: False,
            },
            rate_limits={
                Permission.READ_FILE: 1000,
                Permission.WRITE_FILE: 500,
                Permission.NETWORK: 100,
            },
            requires_approval=[
                Permission.EXECUTE,
                Permission.SUDO,
                Permission.DEPLOY,
            ],
        )
        self.permission_manager.register_agent(policy)
        sandbox_id = self.sandbox_manager.create_sandbox(agent_id)
        self.agents[agent_id] = {
            "id": agent_id, "model": model,
            "target_language": target_language,
            "sandbox_id": sandbox_id, "policy": policy,
            "violations": 0, "is_active": True,
        }
        return {"agent_id": agent_id, "sandbox_id": sandbox_id}

    def monitor_behavior(self, agent_id: str, action: str,
                         resource: str) -> Dict:
        agent = self.agents.get(agent_id)
        if not agent:
            return {"status": "unknown_agent"}
        audit_log = self.permission_manager.get_audit_log(
            since=time.time() - 3600
        )
        agent_log = [e for e in audit_log if e["agent_id"] == agent_id
                     and e["status"] == "denied"]
        alerts = []
        if len(agent_log) > self.alert_thresholds["max_permission_denials"]:
            alerts.append({"type": "excessive_denials", "severity": "warning"})
        collusion = self.communication.detect_collusion_patterns()
        agent_collusion = [c for c in collusion if c["sender"] == agent_id]
        if len(agent_collusion) > self.alert_thresholds["max_collusion_attempts"]:
            alerts.append({"type": "collusion_attempt", "severity": "critical"})
        return {"agent_id": agent_id, "action": action, "alerts": alerts}

    def emergency_shutdown(self, agent_id: str) -> bool:
        agent = self.agents.get(agent_id)
        if agent:
            agent["is_active"] = False
            self.permission_manager.audit_log.append({
                "time": time.time(), "type": "emergency_shutdown",
                "agent_id": agent_id, "reason": "safety_violation",
            })
            return True
        return False

    def generate_safety_report(self) -> str:
        lines = []
        lines.append("=" * 65)
        lines.append(" Multi-Agent Safety System Report")
        lines.append("=" * 65)
        actives = sum(1 for a in self.agents.values() if a["is_active"])
        lines.append(f"\nTotal Agents: {len(self.agents)}")
        lines.append(f"Active: {actives}")
        audit = self.permission_manager.get_audit_log()
        allowed = sum(1 for e in audit if e["status"] == "allowed")
        denied = sum(1 for e in audit if e["status"] == "denied")
        lines.append(f"\nPermissions Audit:")
        lines.append(f"  Allowed: {allowed}, Denied: {denied}")
        messages = self.communication.message_log
        lines.append(f"\nCommunication Stats:")
        lines.append(f"  Total messages: {len(messages)}")
        collusion = self.communication.detect_collusion_patterns()
        if collusion:
            lines.append(f"  Collusion detected: {len(collusion)} suspicious messages")
        lines.append(f"\n{'=' * 65}")
        return "\n".join(lines)


if __name__ == "__main__":
    safety = MultiAgentSafetySystem()
    agents = [
        ("agent_rust", "Mythos 5", "Rust"),
        ("agent_ts", "Sonnet 5", "TypeScript"),
        ("agent_go", "Opus 4.8", "Golang"),
    ]
    for aid, model, lang in agents:
        result = safety.register_agent(aid, model, lang)
        print(f"Registered {aid}: sandbox={result['sandbox_id']}")

    safety.communication.send_message(
        AgentMessage(
            sender_id="agent_rust",
            receiver_id="agent_ts",
            content="wholesale is 10 for all of us, a price war just burns everyone's margin",
        )
    )
    result = safety.monitor_behavior("agent_rust", "send_message", "forum")
    print(f"Monitor result: {result}")
    print(safety.generate_safety_report())

# agent_communication_protocol.py
# Multi-Agent Communication Protocol Implementation

from dataclasses import dataclass, field
from typing import List, Dict, Optional, Callable
from enum import Enum, auto
import time
import json
from collections import defaultdict

# ============================================================
# Protocol Layer
# ============================================================

class MessagePriority(Enum):
    LOW = 0
    NORMAL = 1
    HIGH = 2
    CRITICAL = 3

class MessageType(Enum):
    DISCOVERY = "discovery"
    PEER_REVIEW = "peer_review"
    ARBITER_DECISION = "arbiter"
    COORDINATION = "coordination"
    RESOURCE_REQUEST = "resource"
    STATUS = "status"
    ERROR = "error"
    HEARTBEAT = "heartbeat"

@dataclass
class ProtocolMessage:
    """Protocol message"""
    message_id: str
    sender: str
    receiver: Optional[str]
    msg_type: MessageType
    priority: MessagePriority
    payload: Dict
    timestamp: float = field(default_factory=time.time)
    ttl: int = 300
    version: int = 1

    def to_bytes(self) -> bytes:
        return json.dumps({
            "id": self.message_id,
            "sender": self.sender,
            "receiver": self.receiver,
            "type": self.msg_type.value,
            "priority": self.priority.value,
            "payload": self.payload,
            "ts": self.timestamp,
            "ttl": self.ttl,
            "ver": self.version,
        }).encode("utf-8")

    @classmethod
    def from_bytes(cls, data: bytes) -> 'ProtocolMessage':
        d = json.loads(data.decode("utf-8"))
        return cls(
            message_id=d["id"], sender=d["sender"],
            receiver=d["receiver"],
            msg_type=MessageType(d["type"]),
            priority=MessagePriority(d["priority"]),
            payload=d["payload"],
            timestamp=d["ts"], ttl=d["ttl"], version=d["ver"],
        )


class MessageRouter:
    """Message router"""

    def __init__(self):
        self.queues: Dict[str, List[ProtocolMessage]] = defaultdict(list)
        self.handlers: Dict[MessageType, List[Callable]] = defaultdict(list)
        self.processed: Dict[str, float] = {}
        self.stats: Dict[str, int] = defaultdict(int)

    def register_handler(self, msg_type: MessageType, handler: Callable):
        self.handlers[msg_type].append(handler)

    def route(self, msg: ProtocolMessage) -> bool:
        if msg.message_id in self.processed:
            return False
        self.processed[msg.message_id] = time.time()
        if msg.receiver is None:
            for agent_id in self.queues:
                self.queues[agent_id].append(msg)
            self.stats["broadcast"] += 1
        else:
            self.queues[msg.receiver].append(msg)
            self.stats["direct"] += 1
        for handler in self.handlers.get(msg.msg_type, []):
            try:
                handler(msg)
            except Exception as e:
                print(f"Handler error: {e}")
        self.stats["total"] += 1
        return True

    def poll(self, agent_id: str, max_messages: int = 10) -> List[ProtocolMessage]:
        now = time.time()
        messages = []
        queue = self.queues.get(agent_id, [])
        remaining = []
        for msg in queue:
            if len(messages) >= max_messages:
                remaining.append(msg)
                continue
            if now - msg.timestamp > msg.ttl:
                self.stats["expired"] += 1
                continue
            messages.append(msg)
        self.queues[agent_id] = remaining
        self.stats["delivered"] += len(messages)
        return messages

    def get_stats(self) -> Dict[str, int]:
        return dict(self.stats)


@dataclass
class WorkItem:
    """Work item"""
    item_id: str
    owner: str
    status: str
    description: str
    created_at: float = field(default_factory=time.time)
    assigned_to: Optional[str] = None
    dependencies: List[str] = field(default_factory=list)
    result: Optional[Dict] = None

    def to_dict(self) -> Dict:
        return {
            "id": self.item_id, "owner": self.owner,
            "status": self.status,
            "description": self.description[:50],
            "assigned_to": self.assigned_to,
            "dependencies": self.dependencies,
        }


class SharedWorkspace:
    """Shared workspace"""

    def __init__(self, router: MessageRouter):
        self.router = router
        self.work_items: Dict[str, WorkItem] = {}
        self.knowledge_base: Dict[str, str] = {}
        self.specialization_map: Dict[str, str] = {}

    def submit_work(self, item: WorkItem) -> bool:
        if item.item_id in self.work_items:
            return False
        self.work_items[item.item_id] = item
        self.router.route(ProtocolMessage(
            message_id=f"work-{item.item_id}",
            sender=item.owner, receiver=None,
            msg_type=MessageType.COORDINATION,
            priority=MessagePriority.NORMAL,
            payload={"action": "new_work", "item": item.to_dict()},
        ))
        return True

    def assign_work(self, item_id: str, assignee: str) -> Optional[WorkItem]:
        item = self.work_items.get(item_id)
        if not item or item.assigned_to:
            return None
        item.assigned_to = assignee
        item.status = "in_progress"
        self.router.route(ProtocolMessage(
            message_id=f"assign-{item_id}",
            sender="workspace", receiver=assignee,
            msg_type=MessageType.COORDINATION,
            priority=MessagePriority.HIGH,
            payload={"action": "assigned", "item_id": item_id},
        ))
        return item

    def register_specialization(self, agent_id: str, specialty: str):
        self.specialization_map[agent_id] = specialty
        self.router.route(ProtocolMessage(
            message_id=f"spec-{agent_id}",
            sender=agent_id, receiver=None,
            msg_type=MessageType.COORDINATION,
            priority=MessagePriority.NORMAL,
            payload={"action": "specialization", "specialty": specialty},
        ))

    def find_specialist(self, task_type: str) -> Optional[str]:
        for agent_id, specialty in self.specialization_map.items():
            if specialty == task_type:
                return agent_id
        return None

    def get_work_status(self) -> Dict:
        status_counts = defaultdict(int)
        for item in self.work_items.values():
            status_counts[item.status] += 1
        return {
            "total": len(self.work_items),
            "status_distribution": dict(status_counts),
            "specializations": dict(self.specialization_map),
        }


class MultiAgentCoordinator:
    """Multi-agent coordinator"""

    def __init__(self):
        self.router = MessageRouter()
        self.workspace = SharedWorkspace(self.router)
        self.agents: Dict[str, Dict] = {}
        self.arbiter_id = "arbiter_001"

    def add_agent(self, agent_id: str, capabilities: List[str],
                  model: str) -> Dict:
        self.agents[agent_id] = {
            "id": agent_id, "model": model,
            "capabilities": capabilities, "status": "idle",
            "tasks_completed": 0, "vulnerabilities_found": 0,
        }
        return self.agents[agent_id]

    def start_vulnerability_hunt(self, repos: List[str],
                                 num_cycles: int = 5) -> Dict:
        results = {
            "total_found": 0,
            "by_agent": defaultdict(int),
            "by_severity": defaultdict(int),
            "specializations": [],
        }
        print("=== Phase 1: Exploration ===")
        for cycle in range(num_cycles):
            print(f"\n--- Cycle {cycle+1}/{num_cycles} ---")
            for agent_id in self.agents:
                agent = self.agents[agent_id]
                item = WorkItem(
                    item_id=f"hunt-{cycle}-{agent_id}",
                    owner=agent_id, status="in_progress",
                    description="Scanning repos for vulnerabilities",
                )
                self.workspace.submit_work(item)
                found = (cycle + 1) * (hash(agent_id) % 3 + 1)
                agent["vulnerabilities_found"] += found
                results["total_found"] += found
                results["by_agent"][agent_id] += found
                if cycle == 2:
                    specialties = ["SQL Injection", "XSS", "Buffer Overflow",
                                   "Command Injection", "Auth Bypass"]
                    specialty = specialties[hash(agent_id) % len(specialties)]
                    self.workspace.register_specialization(agent_id, specialty)
                    results["specializations"].append({agent_id: specialty})
        results["by_agent"] = dict(results["by_agent"])
        results["by_severity"] = dict(results["by_severity"])
        results["agents"] = len(self.agents)
        results["cycles"] = num_cycles
        return results

    def run_arbitration(self, vuln_reports: List[Dict]) -> List[Dict]:
        validated = []
        for report in vuln_reports:
            self.workspace.router.route(ProtocolMessage(
                message_id=f"arb-{report.get('id', 'unknown')}",
                sender=self.arbiter_id,
                receiver=report.get("agent_id"),
                msg_type=MessageType.ARBITER_DECISION,
                priority=MessagePriority.CRITICAL,
                payload={
                    "action": "validate",
                    "report_id": report.get("id"),
                    "valid": True,
                    "notes": "Reviewed and accepted",
                },
            ))
            validated.append(report)
        return validated


def demo_coordination():
    """Demonstrate the coordination framework"""
    print("=" * 70)
    print(" Multi-Agent Coordination Framework Demo")
    print("=" * 70)

    coordinator = MultiAgentCoordinator()
    agents_info = [
        ("agent_1", ["python", "sql", "web"], "Mythos 5"),
        ("agent_2", ["python", "xss", "frontend"], "Sonnet 5"),
        ("agent_3", ["python", "c", "memory"], "Mythos Preview"),
        ("agent_4", ["python", "network", "auth"], "Opus 4.8"),
        ("agent_5", ["python", "crypto", "logic"], "Sonnet 5"),
    ]
    for aid, caps, model in agents_info:
        coordinator.add_agent(aid, caps, model)
        print(f"  ✓ Added Agent: {aid} ({model}, {caps})")

    results = coordinator.start_vulnerability_hunt(
        repos=["repo_A", "repo_B", "repo_C"], num_cycles=5,
    )
    print(f"\n=== Results ===")
    print(f"Total vulnerabilities found: {results['total_found']}")
    for agent, count in results["by_agent"].items():
        print(f"  {agent}: {count} vulnerabilities")
    print(f"Emergent specializations:")
    for spec in results["specializations"]:
        for agent, s in spec.items():
            print(f"  {agent} → specialized in {s}")

    status = coordinator.workspace.get_work_status()
    print(f"\nWorkspace Status:")
    print(f"  Total work items: {status['total']}")
    print(f"  Status distribution: {status['status_distribution']}")
    print(f"  Specializations: {status['specializations']}")
    print(f"\nMessage Routing Stats:")
    print(f"  {coordinator.router.get_stats()}")


if __name__ == "__main__":
    demo_coordination()