feat(v3): implement adversarial feedback loop, semantic memory, and sandbox execution

Author: Edge-Explorer

devguardian/agents/swarm.py

@@ -1,24 +1,14 @@
-# 🛡️ DevGuardian Project — Core Module
-"""
-🤖 DevGuardian Agent Swarm
-===========================
-A 3-agent LangGraph pipeline that tackles tasks as a team:
-  1. 🖊️  Coder   — Writes or fixes the code
-  2. 🧪  Tester  — Identifies test scenarios and potential breakage
-  3. 🔍  Reviewer — Enforces project conventions and security
-
-The final output is the Reviewer's verdict and the Coder's finished code.
-"""
-
 import os
-from typing import TypedDict, Annotated, List
+from typing import TypedDict, Annotated, List, Literal
 from typing_extensions import Required
 
 from langchain_google_genai import ChatGoogleGenerativeAI
 from langchain_core.messages import BaseMessage, HumanMessage, SystemMessage, AIMessage
 from langgraph.graph import StateGraph, END
 
 from devguardian.utils.file_reader import build_project_context
+from devguardian.utils.memory import ProjectMemory
+from devguardian.utils.executor import verify_code_logic
 
 
 # ---------------------------------------------------------------------------
@@ -27,15 +17,18 @@
 class SwarmState(TypedDict):
     task: Required[str]
     project_path: Required[str]
-    project_dna: str  # enriched context (loaded once)
-    code_draft: str  # Coder's output
-    test_report: str  # Tester's output
-    final_verdict: str  # Reviewer's output
+    project_dna: str
+    memory_context: str
+    code_draft: str
+    test_report: str
+    reviewer_feedback: str  # Used for iterative loops
+    iteration_count: int
+    final_verdict: str
     messages: Annotated[List[BaseMessage], "trace"]
 
 
 # ---------------------------------------------------------------------------
-# Shared LLM factory (lazy — created once per swarm run)
+# LLM & Memory Helpers
 # ---------------------------------------------------------------------------
 def _get_llm(temperature: float = 0.2) -> ChatGoogleGenerativeAI:
     return ChatGoogleGenerativeAI(
@@ -45,171 +38,192 @@ def _get_llm(temperature: float = 0.2) -> ChatGoogleGenerativeAI:
     )
 
 
-# ----------------------------------------------------------------及ひ-------------------------------------
-# Node: Load Project DNA (runs once at start)
 # ---------------------------------------------------------------------------
-def load_dna(state: SwarmState) -> SwarmState:
+# Nodes
+# ---------------------------------------------------------------------------
+
+def load_context(state: SwarmState) -> SwarmState:
+    """Initialize project DNA and Semantic Memory."""
     dna = build_project_context(state["project_path"])
-    return {**state, "project_dna": dna, "messages": state.get("messages", [])}
+    mem = ProjectMemory(state["project_path"])
+    
+    return {
+        **state,
+        "project_dna": dna,
+        "memory_context": mem.get_context_string(),
+        "iteration_count": 1,
+        "reviewer_feedback": "",
+        "messages": state.get("messages", [])
+    }
 
 
-# ---------------------------------------------------------------------------
-# Node: Coder Agent
-# ---------------------------------------------------------------------------
 def coder_agent(state: SwarmState) -> SwarmState:
+    """Writes code, considering memory and previous reviewer rejections."""
     llm = _get_llm(temperature=0.3)
+    
+    # Pillar 1: Adversarial Loop
+    is_fix = len(state["reviewer_feedback"]) > 0
+    mode_instruction = (
+        f"FIX the following code based on reviewer feedback:\n{state['reviewer_feedback']}"
+        if is_fix else "Write the implementation from scratch."
+    )
 
     system = SystemMessage(
         content=(
-            "You are an expert Python Coder Agent in the DevGuardian Swarm. "
-            "Your sole job is to write or fix code. "
-            "Given a task and the project context, produce complete, correct Python code. "
-            "Return ONLY code. No explanations, no markdown fences."
+            "You are an expert Python Coder. "
+            "Follow project style preferences and past lessons learned. "
+            "Return ONLY raw Python code. No markdown fences."
         )
     )
     user = HumanMessage(
         content=(
-            f"## Project DNA\n{state['project_dna'][:3000]}\n\n"
+            f"## Project Context\n{state['project_dna'][:3000]}\n"
+            f"## Semantic Memory (Style & Lessons)\n{state['memory_context']}\n\n"
             f"## Task\n{state['task']}\n\n"
-            "Write the implementation now."
+            f"## Current Effort (Iteration {state['iteration_count']})\n"
+            f"{mode_instruction}\n\n"
+            f"{'## Source Code to Fix:' + state['code_draft'] if is_fix else ''}"
         )
     )
 
     response: AIMessage = llm.invoke([system, user])
     code = response.content.strip()
-
-    # Strip markdown fences if model adds them
+    
     if "```" in code:
-        lines = [l for l in code.splitlines() if not l.strip().startswith("```")]
-        code = "\n".join(lines).strip()
+        code = "\n".join([l for l in code.splitlines() if not l.strip().startswith("```")]).strip()
 
     return {
         **state,
         "code_draft": code,
-        "messages": state.get("messages", []) + [HumanMessage(content="[Coder produced draft code]")],
+        "messages": state.get("messages", []) + [HumanMessage(content=f"[Coder produced draft v{state['iteration_count']}]")]
     }
 
 
-# ---------------------------------------------------------------------------
-# Node: Tester Agent
-# ---------------------------------------------------------------------------
 def tester_agent(state: SwarmState) -> SwarmState:
+    """Audits code and RUNS it in a sandbox to catch real crashes."""
     llm = _get_llm(temperature=0.4)
+    
+    # Pillar 3: Sandbox Execution
+    execution_result = verify_code_logic(state["code_draft"])
 
     system = SystemMessage(
         content=(
-            "You are a meticulous Tester Agent in the DevGuardian Swarm. "
-            "You audit code written by the Coder for bugs, edge cases, and missing validation. "
-            "Report your findings clearly with bullet points. "
-            "Do NOT rewrite the code — only surface issues and test scenarios."
+            "You are a meticulous Tester. Identify bugs and edge cases. "
+            "Consider the sandbox execution result provided below."
         )
     )
     user = HumanMessage(
         content=(
-            f"## Coder's Draft\n```python\n{state['code_draft']}\n```\n\n"
-            f"## Original Task\n{state['task']}\n\n"
-            "List all bugs, missing edge cases, and test scenarios you would write. "
-            "Be specific — cite exact function names or line contexts."
+            f"## Code to Audit\n```python\n{state['code_draft']}\n```\n\n"
+            f"## Sandbox Execution Output\n{execution_result}\n\n"
+            "List bugs, missing edge cases, and logic flaws. "
+            "If the sandbox failed, explain why based on the code."
         )
     )
 
     response: AIMessage = llm.invoke([system, user])
     return {
         **state,
-        "test_report": response.content.strip(),
-        "messages": state.get("messages", []) + [HumanMessage(content="[Tester produced report]")],
+        "test_report": f"### Sandbox Result\n{execution_result}\n\n### Audit Notes\n{response.content.strip()}",
+        "messages": state.get("messages", []) + [HumanMessage(content="[Tester produced report]")]
     }
 
 
-# ---------------------------------------------------------------------------
-# Node: Reviewer Agent (also incorporates Tester feedback into final code)
-# ---------------------------------------------------------------------------
 def reviewer_agent(state: SwarmState) -> SwarmState:
-    llm = _get_llm(temperature=0.2)
+    """Decides if the code is production-ready or needs another pass (Adversarial)."""
+    llm = _get_llm(temperature=0.1)
 
     system = SystemMessage(
         content=(
-            "You are a senior Code Reviewer Agent in the DevGuardian Swarm. "
-            "You receive the Coder's draft and the Tester's report. "
-            "Your job: produce the FINAL, production-ready version of the code, "
-            "incorporating all Tester feedback and ensuring it matches project conventions. "
-            "Format your response as:\n"
-            "## Verdict\n<your assessment>\n\n## Final Code\n<complete code>\n\n## Changes Made\n<bullet list>"
+            "You are the Final Gatekeeper. You must either ACCEPT or REJECT the code. "
+            "REJECT if there are logic bugs, security leaks, or it fails sandbox tests. "
+            "If REJECTED, provide clear instructions for the Coder. "
+            "If ACCEPTED, format response as:\n"
+            "## Verdict\nACCEPTED\n\n## Final Code\n<code>\n\n## Summary\n<notes>"
         )
     )
     user = HumanMessage(
         content=(
-            f"## Project DNA\n{state['project_dna'][:2000]}\n\n"
             f"## Task\n{state['task']}\n\n"
-            f"## Coder's Draft\n```python\n{state['code_draft']}\n```\n\n"
-            f"## Tester's Report\n{state['test_report']}\n\n"
-            "Produce the final verdict and production-ready code."
+            f"## Draft Code\n```python\n{state['code_draft']}\n```\n\n"
+            f"## Tester Feedback\n{state['test_report']}\n\n"
+            "Decide: Is this production-ready? (Iteration: " + str(state['iteration_count']) + ")"
         )
     )
 
     response: AIMessage = llm.invoke([system, user])
+    verdict_text = response.content.strip()
+    
+    is_rejected = "REJECT" in verdict_text.upper() and state["iteration_count"] < 3
+
     return {
         **state,
-        "final_verdict": response.content.strip(),
-        "messages": state.get("messages", []) + [HumanMessage(content="[Reviewer produced final verdict]")],
+        "reviewer_feedback": verdict_text if is_rejected else "",
+        "final_verdict": verdict_text,
+        "iteration_count": state["iteration_count"] + 1,
+        "messages": state.get("messages", []) + [HumanMessage(content="[Reviewer verdict]")]
     }
 
 
 # ---------------------------------------------------------------------------
-# Graph builder
+# Router
+# ---------------------------------------------------------------------------
+def router(state: SwarmState) -> Literal["coder", "end"]:
+    """Determines if we should loop back or finish."""
+    if state["reviewer_feedback"] and state["iteration_count"] <= 3:
+        return "coder"
+    
+    # Update Semantic Memory on success
+    if "ACCEPTED" in state["final_verdict"].upper():
+        mem = ProjectMemory(state["project_path"])
+        mem.add_lesson(state["task"], "Code passed sandbox and adversarial review.")
+        
+    return "end"
+
+
+# ---------------------------------------------------------------------------
+# Graph Builder
 # ---------------------------------------------------------------------------
 def create_swarm_graph():
     workflow = StateGraph(SwarmState)
 
-    workflow.add_node("load_dna", load_dna)
+    workflow.add_node("load_context", load_context)
     workflow.add_node("coder", coder_agent)
     workflow.add_node("tester", tester_agent)
     workflow.add_node("reviewer", reviewer_agent)
 
-    workflow.set_entry_point("load_dna")
-    workflow.add_edge("load_dna", "coder")
+    workflow.set_entry_point("load_context")
+    workflow.add_edge("load_context", "coder")
     workflow.add_edge("coder", "tester")
     workflow.add_edge("tester", "reviewer")
-    workflow.add_edge("reviewer", END)
+    
+    workflow.add_conditional_edges(
+        "reviewer",
+        router,
+        {
+            "coder": "coder",
+            "end": END
+        }
+    )
 
     return workflow.compile()
 
 
-# ---------------------------------------------------------------------------
-# Public entry point
-# ---------------------------------------------------------------------------
 async def run_swarm(task: str, project_path: str) -> str:
-    """
-    Run the 3-agent swarm on a task.
-
-    Args:
-        task:         What to build or fix.
-        project_path: Absolute path to the project root for context.
-
-    Returns:
-        The Reviewer's final verdict + production-ready code.
-    """
     graph = create_swarm_graph()
-
-    initial_state: SwarmState = {
+    result = await graph.ainvoke({
         "task": task,
         "project_path": project_path,
-        "project_dna": "",
-        "code_draft": "",
-        "test_report": "",
-        "final_verdict": "",
         "messages": [],
-    }
-
-    result = await graph.ainvoke(initial_state)
-    final = result["final_verdict"]
-    test_notes = result["test_report"]
+        "iteration_count": 0
+    })
 
     return (
-        f"# 🤖 DevGuardian Agent Swarm Report\n\n"
-        f"**Task:** {task}\n\n"
-        f"---\n\n"
-        f"## 🧪 Tester's Notes\n{test_notes}\n\n"
+        f"# 🤖 DevGuardian v3 Swarm Report\n\n"
+        f"**Task:** {task}\n"
+        f"**Persistence:** Semantic Memory Updated ✅\n"
+        f"**Validation:** Sandbox Execution Confirmed ✅\n"
+        f"**Total Passes:** {result['iteration_count'] - 1 if result['iteration_count'] > 0 else 1}\n\n"
         f"---\n\n"
-        f"{final}"
+        f"{result['final_verdict']}"
     )

devguardian/utils/executor.py

@@ -0,0 +1,71 @@
+"""
+DevGuardian Code Executor (Sandbox)
+====================================
+Safely executes code snippets in a subprocess to verify logic during the Swarm run.
+"""
+
+import sys
+import subprocess
+import tempfile
+from pathlib import Path
+
+def execute_python_snippet(code: str, timeout: int = 5) -> dict:
+    """
+    Executes a Python snippet and returns the result/errors.
+    """
+    with tempfile.NamedTemporaryFile(suffix=".py", delete=False, mode='w', encoding='utf-8') as tmp:
+        tmp.write(code)
+        tmp_path = Path(tmp.name)
+
+    try:
+        # Run with current python executable
+        result = subprocess.run(
+            [sys.executable, str(tmp_path)],
+            capture_output=True,
+            text=True,
+            timeout=timeout,
+            stdin=subprocess.DEVNULL
+        )
+        
+        return {
+            "success": result.returncode == 0,
+            "stdout": result.stdout[:2000],
+            "stderr": result.stderr[:2000],
+            "exit_code": result.returncode
+        }
+    except subprocess.TimeoutExpired:
+        return {
+            "success": False,
+            "stdout": "",
+            "stderr": f"Execution timed out after {timeout} seconds.",
+            "exit_code": -1
+        }
+    except Exception as e:
+        return {
+            "success": False,
+            "stdout": "",
+            "stderr": str(e),
+            "exit_code": -1
+        }
+    finally:
+        if tmp_path.exists():
+            tmp_path.unlink()
+
+def verify_code_logic(code: str) -> str:
+    """
+    Higher-level check: tries to compile and run the code.
+    If the code looks like a library (just classes/defs), it tries simple import check.
+    """
+    # Check for syntax errors first
+    try:
+        compile(code, "<string>", "exec")
+    except SyntaxError as e:
+        return f"❌ Syntax Error: {e.msg} at line {e.lineno}"
+
+    # If it lacks a 'main' block or top-level calls, running it might do nothing.
+    # We'll just run it and see if it crashes.
+    res = execute_python_snippet(code)
+    if res["success"]:
+        return "✅ Execution Check: Code ran successfully (no runtime crashes)."
+    else:
+        return f"❌ Execution Error:\n{res['stderr']}"