# Multi-Agent Swarm Orchestrator & Autonomous Workflow OS ==================================================================== MODULE 1: SWARM TOPOLOGY, STATE SCHEMAS & COMMUNICATION CONTRACTS ==================================================================== - Swarm Architectural Blueprint: * Swarm Name: SecurePatch Swarm * Topology: Hierarchical Supervisor with Adversarial Reflection Loop * Mission: Autonomously ingest GitHub Pull Requests, execute static security analysis (Semgrep), generate minimal secure code diffs, verify changes with a security auditor agent, and request human lead approval prior to submitting PR reviews. * Agent Roster: 1. `Supervisor`: Orchestrates triage, routes code to specialists, and manages flow convergence. 2. `StaticAnalyzer`: Executes SAST tooling and aggregates CWE vulnerabilities. 3. `RemediationEngineer`: Writes surgical, bug-free code patches fixing detected vulnerabilities. 4. `SecurityAuditor`: Adversarial critic verifying patch correctness, regressions, and zero side-effects. - Production Pydantic v2 & TypedDict State Schema: ```python from typing import Annotated, List, Dict, Any, Optional, Literal from typing_extensions import TypedDict from pydantic import BaseModel, Field from langgraph.graph.message import add_messages from langchain_core.messages import BaseMessage class VulnerabilityReport(BaseModel): cwe_id: str = Field(description="Common Weakness Enumeration ID, e.g., CWE-89") severity: Literal["LOW", "MEDIUM", "HIGH", "CRITICAL"] file_path: str line_number: int description: str recommended_fix: str class PatchProposal(BaseModel): file_path: str original_code: str patched_code: str explanation: str class AuditResult(BaseModel): is_approved: bool security_score: int = Field(ge=0, le=100) identified_issues: List[str] = Field(default_factory=list) rejection_reason: Optional[str] = None class AgentState(TypedDict): messages: Annotated[List[BaseMessage], add_messages] repo_name: str pr_number: int target_files: List[str] vulnerabilities: List[Dict[str, Any]] proposed_patches: List[Dict[str, Any]] audit_verdict: Optional[Dict[str, Any]] iteration_count: int next_agent: str human_approved: Optional[bool] { "recipient": "RemediationEngineer", "action": "GENERATE_PATCH", "payload": { "vulnerability_count": 2, "target_files": ["src/auth/jwt.py"], "rejection_context": "Previous patch introduced a null pointer on line 42" } } import operator from typing import Literal from langchain_core.messages import HumanMessage, SystemMessage from langchain_anthropic import ChatAnthropic from langgraph.graph import StateGraph, END from langgraph.checkpoint.memory import MemorySaver from pydantic import BaseModel, Field llm = ChatAnthropic(model="claude-3-5-sonnet-20241022", temperature=0.0) class SupervisorRouter(BaseModel): next_agent: Literal["StaticAnalyzer", "RemediationEngineer", "SecurityAuditor", "HumanApproval", "FINISH"] reasoning: str def supervisor_node(state: AgentState) -> Dict[str, Any]: if state.get("iteration_count", 0) >= 10: return {"next_agent": "FINISH", "messages": [SystemMessage(content="Circuit breaker tripped: Max iterations reached.")]} system_prompt = ( "You are the Lead Security Supervisor. Review current state and route to the single appropriate agent: " "1. StaticAnalyzer: If vulnerabilities are empty. " "2. RemediationEngineer: If vulnerabilities exist but proposed_patches are empty or audit failed. " "3. SecurityAuditor: If proposed_patches exist and audit_verdict is empty. " "4. HumanApproval: If audit_verdict.is_approved is True. " "5. FINISH: If human_approved is True or no action needed." ) structured_llm = llm.with_structured_output(SupervisorRouter) decision = structured_llm.invoke([SystemMessage(content=system_prompt)] + state["messages"]) return { "next_agent": decision.next_agent, "iteration_count": state.get("iteration_count", 0) + 1, "messages": [SystemMessage(content=f"Supervisor assigned: {decision.next_agent}. Reasoning: {decision.reasoning}")] } def route_decision(state: AgentState) -> str: return state["next_agent"] workflow = StateGraph(AgentState) workflow.add_node("Supervisor", supervisor_node) # Additional nodes added in Module 5 import json import subprocess from langchain_core.tools import tool class SemgrepScanInput(BaseModel): target_path: str = Field(description="Relative filesystem path to run SAST scan on") rule_set: str = Field(default="p/security-audit", description="Semgrep ruleset") @tool(args_schema=SemgrepScanInput) def run_semgrep_scan(target_path: str, rule_set: str = "p/security-audit") -> str: """Executes a local Semgrep security scan and returns structured findings.""" try: cmd = ["semgrep", "scan", "--config", rule_set, "--json", target_path] result = subprocess.run(cmd, capture_output=True, text=True, timeout=45) return result.stdout if result.stdout else json.dumps({"results": []}) except subprocess.TimeoutExpired: return json.dumps({"error": "Semgrep scan timed out after 45 seconds"}) except Exception as e: return json.dumps({"error": str(e)}) # Human lead supplies approval decision: resume_payload = {"human_approved": True} # Or False with human feedback comment from typing import Dict, Any from langchain_core.messages import SystemMessage, HumanMessage from langgraph.graph import StateGraph, END from langgraph.checkpoint.memory import MemorySaver def static_analyzer_node(state: AgentState) -> Dict[str, Any]: # Simulated node execution adhering to schema mock_vuln = { "cwe_id": "CWE-89", "severity": "HIGH", "file_path": "src/auth/db.py", "line_number": 18, "description": "SQL Injection vulnerability via raw string format.", "recommended_fix": "Use parameterized queries." } return {"vulnerabilities": [mock_vuln]} def remediation_node(state: AgentState) -> Dict[str, Any]: mock_patch = { "file_path": "src/auth/db.py", "original_code": "cursor.execute(f'SELECT * FROM users WHERE id = {user_id}')", "patched_code": "cursor.execute('SELECT * FROM users WHERE id = %s', (user_id,))", "explanation": "Replaced f-string with parameterized query argument." } return {"proposed_patches": [mock_patch]} def security_auditor_node(state: AgentState) -> Dict[str, Any]: # Evaluates proposed patches return { "audit_verdict": { "is_approved": True, "security_score": 98, "identified_issues": [] } } def human_approval_node(state: AgentState) -> Dict[str, Any]: # Execution halts before this node via interrupt_before return {"messages": [SystemMessage(content="Human approved patch deployment.")]} builder = StateGraph(AgentState) builder.add_node("Supervisor", supervisor_node) builder.add_node("StaticAnalyzer", static_analyzer_node) builder.add_node("RemediationEngineer", remediation_node) builder.add_node("SecurityAuditor", security_auditor_node) builder.add_node("HumanApproval", human_approval_node) builder.set_entry_point("Supervisor") builder.add_conditional_edges("Supervisor", route_decision, { "StaticAnalyzer": "StaticAnalyzer", "RemediationEngineer": "RemediationEngineer", "SecurityAuditor": "SecurityAuditor", "HumanApproval": "HumanApproval", "FINISH": END }) builder.add_edge("StaticAnalyzer", "Supervisor") builder.add_edge("RemediationEngineer", "Supervisor") builder.add_edge("SecurityAuditor", "Supervisor") builder.add_edge("HumanApproval", "Supervisor") memory = MemorySaver() app = builder.compile(checkpointer=memory, interrupt_before=["HumanApproval"]) import pytest from langchain_core.messages import HumanMessage def test_swarm_initialization(): initial_state = { "messages": [HumanMessage(content="Audit PR #104")], "repo_name": "acme/backend", "pr_number": 104, "target_files": ["src/auth/db.py"], "vulnerabilities": [], "proposed_patches": [], "audit_verdict": None, "iteration_count": 0, "next_agent": "", "human_approved": None } config = {"configurable": {"thread_id": "test_thread_1"}} # Run until interrupt result = app.invoke(initial_state, config=config) state_history = app.get_state(config) assert state_history.next == ("HumanApproval",) assert len(state_history.values["vulnerabilities"]) > 0 assert len(state_history.values["proposed_patches"]) > 0 assert state_history.values["audit_verdict"]["is_approved"] is True def test_human_approval_resume(): config = {"configurable": {"thread_id": "test_thread_1"}} # Resume paused graph app.update_state(config, {"human_approved": True}, as_node="HumanApproval") final_state = app.invoke(None, config=config) assert final_state["next_agent"] == "FINISH"