Why One Agent Isn't Enough

Input: "Add rate limiting to API endpoints"
Output:
  1. Read existing middleware
  2. Create rate_limit.py with token bucket algorithm
  3. Add middleware to app.py
  4. Create tests/test_rate_limit.py
  5. Update documentation

Input: Plan from Planner
Actions:
  - Read relevant files
  - Write new code
  - Edit existing files
  - Run basic syntax checks
Output: Code changes ready for testing

Input: Code changes from Coder
Actions:
  - Run pytest/vitest
  - Analyze failures
  - Report which tests fail and why
Output: Test results (pass/fail + details)

Input: Code changes
Checks:
  - Input validation
  - SQL injection
  - XSS vulnerabilities
  - Hard-coded secrets
  - Dependency vulnerabilities
Output: Security report (issues found or clear)

Input: All previous results
Checks:
  - Code style consistency
  - Documentation completeness
  - Test coverage
  - No regressions
  - PR description quality
Output: Approved or changes requested

"""Multi-Agent Software Development Simulation — Safe demo."""

from dataclasses import dataclass, field
from typing import Callable
from enum import Enum


class AgentRole(Enum):
    PLANNER = "Planner"
    CODER = "Coder"
    TESTER = "Tester"
    SECURITY = "Security"
    REVIEWER = "Reviewer"


class TaskStatus(Enum):
    PENDING = "pending"
    IN_PROGRESS = "in_progress"
    PASSED = "passed"
    FAILED = "failed"
    RETRY = "retry"


@dataclass
class Task:
    description: str
    status: TaskStatus = TaskStatus.PENDING
    plan: list[str] = field(default_factory=list)
    code_changes: list[str] = field(default_factory=list)
    test_results: dict = field(default_factory=dict)
    security_issues: list[str] = field(default_factory=list)
    review_comments: list[str] = field(default_factory=list)
    retry_count: int = 0


@dataclass
class AgentResult:
    agent: AgentRole
    success: bool
    output: str
    issues: list[str] = field(default_factory=list)


class MultiAgentOrchestrator:
    """Simulates a multi-agent development workflow."""

    def __init__(self, max_retries: int = 3):
        self.max_retries = max_retries
        self.agents: dict[AgentRole, Callable] = {
            AgentRole.PLANNER: self._planner_agent,
            AgentRole.CODER: self._coder_agent,
            AgentRole.TESTER: self._tester_agent,
            AgentRole.SECURITY: self._security_agent,
            AgentRole.REVIEWER: self._reviewer_agent,
        }
        self.history: list[AgentResult] = []

    def execute(self, task: Task) -> Task:
        """Run the full multi-agent pipeline."""
        print(f"\n{'='*50}")
        print(f"Task: {task.description}")
        print(f"{'='*50}")

        # Step 1: Plan
        result = self._run_agent(AgentRole.PLANNER, task)
        if not result.success:
            return task

        # Step 2: Implement
        result = self._run_agent(AgentRole.CODER, task)
        if not result.success:
            return task

        # Step 3: Test (with retry loop)
        while task.retry_count < self.max_retries:
            result = self._run_agent(AgentRole.TESTER, task)
            if result.success:
                break
            task.retry_count += 1
            print(f"  ↻ Retry {task.retry_count}/{self.max_retries}")
            self._run_agent(AgentRole.CODER, task)

        if task.status == TaskStatus.FAILED:
            return task

        # Step 4: Security check
        result = self._run_agent(AgentRole.SECURITY, task)
        if not result.success:
            task.retry_count += 1
            if task.retry_count < self.max_retries:
                self._run_agent(AgentRole.CODER, task)
                return self.execute(task)  # Restart pipeline

        # Step 5: Review
        result = self._run_agent(AgentRole.REVIEWER, task)
        return task

    def _run_agent(self, role: AgentRole, task: Task) -> AgentResult:
        """Run a single agent and record the result."""
        agent_fn = self.agents[role]
        result = agent_fn(task)
        self.history.append(result)

        status = "✅" if result.success else "❌"
        print(f"{status} {role.value}: {result.output}")
        if result.issues:
            for issue in result.issues:
                print(f"   ⚠️  {issue}")

        return result

    # ── Agent Implementations (mock) ──────────────────────

    def _planner_agent(self, task: Task) -> AgentResult:
        task.plan = [
            "Read existing middleware structure",
            "Implement rate limiting logic",
            "Add middleware to application",
            "Write unit tests",
        ]
        task.status = TaskStatus.IN_PROGRESS
        return AgentResult(
            agent=AgentRole.PLANNER,
            success=True,
            output=f"Created plan with {len(task.plan)} steps",
        )

    def _coder_agent(self, task: Task) -> AgentResult:
        task.code_changes = [
            "Created rate_limit.py",
            "Modified app.py to add middleware",
            "Created tests/test_rate_limit.py",
        ]
        return AgentResult(
            agent=AgentRole.CODER,
            success=True,
            output=f"Implemented {len(task.code_changes)} changes",
        )

    def _tester_agent(self, task: Task) -> AgentResult:
        # Simulate: first attempt fails, second passes
        if task.retry_count == 0 and not task.test_results:
            task.test_results = {"passed": 3, "failed": 1}
            task.status = TaskStatus.FAILED
            return AgentResult(
                agent=AgentRole.TESTER,
                success=False,
                output="3 passed, 1 failed",
                issues=["test_rate_limit_exceeded failed"],
            )
        task.test_results = {"passed": 4, "failed": 0}
        task.status = TaskStatus.PASSED
        return AgentResult(
            agent=AgentRole.TESTER,
            success=True,
            output="4 passed, 0 failed",
        )

    def _security_agent(self, task: Task) -> AgentResult:
        task.security_issues = []
        return AgentResult(
            agent=AgentRole.SECURITY,
            success=True,
            output="No security issues found",
        )

    def _reviewer_agent(self, task: Task) -> AgentResult:
        task.review_comments = [
            "Code follows project style",
            "Tests cover edge cases",
            "Documentation updated",
        ]
        return AgentResult(
            agent=AgentRole.REVIEWER,
            success=True,
            output="Approved — ready for merge",
        )


# ── Run Simulation ────────────────────────────────────────

if __name__ == "__main__":
    orchestrator = MultiAgentOrchestrator(max_retries=3)

    task = Task(
        description="Add rate limiting to API endpoints"
    )

    final = orchestrator.execute(task)

    print(f"\n{'='*50}")
    print("Final Status:")
    print(f"  Status: {final.status.value}")
    print(f"  Plan steps: {len(final.plan)}")
    print(f"  Code changes: {len(final.code_changes)}")
    print(f"  Test results: {final.test_results}")
    print(f"  Security issues: {len(final.security_issues)}")
    print(f"  Review comments: {len(final.review_comments)}")
    print(f"  Retries: {final.retry_count}")
    print(f"{'='*50}")