Add performance and quality audit agents

README.md

@@ -7,7 +7,7 @@ AI-powered multi-agent code auditing for GitHub repositories. Paste a public Git
 SwarmAudit currently runs with a mock-first LLM interface so the demo is not blocked by ROCm, vLLM, or AMD MI300X setup. The current graph is:
 
 ```text
-GitHub URL -> Crawler -> Chunker -> [Security Agent + Performance Agent] -> Synthesizer -> Report
+GitHub URL -> Crawler -> Chunker -> [Security Agent + Performance Agent + Quality Agent] -> Synthesizer -> Report
 ```
 
 ## Quick Start
@@ -62,6 +62,7 @@ Each finding includes:
 
 - Security Agent: flags hardcoded secrets, disabled TLS verification, and dynamic code execution.
 - Performance Agent: flags HTTP calls without timeouts, blocking sleep inside async functions, nested loops, file reads in loops, and synchronous Node.js filesystem calls.
+- Quality Agent: flags long functions, high branch density, large source sections, unresolved TODO/FIXME/HACK comments, and very short symbol names.
 - Synthesizer Agent: deduplicates findings, sorts by severity, and builds the final report.
 
 ## Tests

app/agents/graph.py

@@ -5,6 +5,7 @@ from typing import Annotated, TypedDict
 from langgraph.graph import END, StateGraph
 
 from app.agents.performance_agent import PerformanceAgent
+from app.agents.quality_agent import QualityAgent
 from app.agents.security_agent import SecurityAgent
 from app.agents.synthesizer_agent import SynthesizerAgent
 from app.config import Settings, get_settings
@@ -20,6 +21,7 @@ class AuditState(TypedDict, total=False):
     chunks: list[CodeChunk]
     security_output: AgentOutput
     performance_output: AgentOutput
+    quality_output: AgentOutput
     report: AuditReport
     progress: Annotated[list[str], add]
 
@@ -32,6 +34,7 @@ class AuditGraph:
         self.llm_client = LLMClient(self.settings)
         self.security_agent = SecurityAgent(self.llm_client)
         self.performance_agent = PerformanceAgent()
+        self.quality_agent = QualityAgent()
         self.synthesizer = SynthesizerAgent()
         self.graph = self._build_graph()
 
@@ -41,12 +44,14 @@ class AuditGraph:
         graph.add_node("chunk", self._chunk)
         graph.add_node("security", self._security)
         graph.add_node("performance", self._performance)
+        graph.add_node("quality", self._quality)
         graph.add_node("synthesize", self._synthesize)
         graph.set_entry_point("crawl")
         graph.add_edge("crawl", "chunk")
         graph.add_edge("chunk", "security")
         graph.add_edge("chunk", "performance")
-        graph.add_edge(["security", "performance"], "synthesize")
+        graph.add_edge("chunk", "quality")
+        graph.add_edge(["security", "performance", "quality"], "synthesize")
         graph.add_edge("synthesize", END)
         return graph.compile()
 
@@ -73,8 +78,12 @@ class AuditGraph:
             performance_output = await self.performance_agent.analyze(chunks)
             yield f"Performance Agent: found {len(performance_output.findings)} findings."
 
+            yield "Quality Agent: scanning maintainability signals..."
+            quality_output = await self.quality_agent.analyze(chunks)
+            yield f"Quality Agent: found {len(quality_output.findings)} findings."
+
             yield "Synthesizer Agent: ranking findings and formatting report..."
-            report = await self.synthesizer.synthesize(repo, [security_output, performance_output])
+            report = await self.synthesizer.synthesize(repo, [security_output, performance_output, quality_output])
             yield "Synthesizer Agent: final report generated."
             yield report
         finally:
@@ -96,10 +105,14 @@ class AuditGraph:
         output = await self.performance_agent.analyze(state["chunks"])
         return {"performance_output": output, "progress": [f"Performance Agent: found {len(output.findings)} findings."]}
 
+    async def _quality(self, state: AuditState) -> AuditState:
+        output = await self.quality_agent.analyze(state["chunks"])
+        return {"quality_output": output, "progress": [f"Quality Agent: found {len(output.findings)} findings."]}
+
     async def _synthesize(self, state: AuditState) -> AuditState:
         report = await self.synthesizer.synthesize(
             state["repo"],
-            [state["security_output"], state["performance_output"]],
+            [state["security_output"], state["performance_output"], state["quality_output"]],
         )
         self.crawler.cleanup(state["repo"])
         return {"report": report, "progress": ["Synthesizer Agent: final report generated."]}

app/agents/quality_agent.py

@@ -1,8 +1,195 @@
-from app.schemas import AgentOutput, CodeChunk
+import re
+
+from app.schemas import AgentOutput, CodeChunk, Finding, Severity
+
+
+PYTHON_DEF = re.compile(r"^\s*(async\s+def|def|class)\s+([A-Za-z_][A-Za-z0-9_]*)")
+PYTHON_BRANCH = re.compile(r"^\s*(if|elif|for|while|except|with)\b")
+JS_FUNCTION = re.compile(r"^\s*(function\s+[A-Za-z_$][\w$]*|(?:const|let|var)\s+[A-Za-z_$][\w$]*\s*=\s*(?:async\s*)?\()")
+JS_BRANCH = re.compile(r"^\s*(if|else\s+if|for|while|switch|catch)\b")
+TODO_COMMENT = re.compile(r"(?i)\b(TODO|FIXME|HACK)\b")
+
+
+MAX_CHUNK_LINES = 300
+MAX_FUNCTION_LINES = 80
+MAX_BRANCHES_PER_CHUNK = 25
+MIN_MEANINGFUL_NAME_LENGTH = 3
 
 
 class QualityAgent:
     name = "Quality Agent"
 
     async def analyze(self, chunks: list[CodeChunk]) -> AgentOutput:
-        return AgentOutput(agent_name=self.name, findings=[], metadata={"chunks_scanned": len(chunks)})
+        findings: list[Finding] = []
+        for chunk in chunks:
+            findings.extend(self._scan_chunk(chunk))
+
+        return AgentOutput(
+            agent_name=self.name,
+            findings=findings,
+            metadata={"chunks_scanned": len(chunks), "mode": "static-rules"},
+        )
+
+    def _scan_chunk(self, chunk: CodeChunk) -> list[Finding]:
+        findings: list[Finding] = []
+        lines = chunk.content.splitlines()
+
+        findings.extend(self._check_large_chunk(chunk, lines))
+        findings.extend(self._check_long_functions(chunk, lines))
+        findings.extend(self._check_branch_density(chunk, lines))
+        findings.extend(self._check_todo_comments(chunk, lines))
+        findings.extend(self._check_short_names(chunk, lines))
+
+        return findings
+
+    def _check_large_chunk(self, chunk: CodeChunk, lines: list[str]) -> list[Finding]:
+        if len(lines) <= MAX_CHUNK_LINES:
+            return []
+
+        return [
+            self._finding(
+                "Large source file section",
+                Severity.low,
+                chunk,
+                chunk.line_start,
+                chunk.line_end,
+                "This source section is large enough to make review, testing, and future changes harder.",
+                "Split unrelated responsibilities into smaller modules or focused helper functions.",
+            )
+        ]
+
+    def _check_long_functions(self, chunk: CodeChunk, lines: list[str]) -> list[Finding]:
+        findings: list[Finding] = []
+        active_start: int | None = None
+        active_name: str | None = None
+        active_indent = 0
+
+        for index, line in enumerate(lines):
+            if not line.strip():
+                continue
+
+            match = self._definition_match(line)
+            indent = self._indent_width(line)
+
+            if active_start is not None and indent <= active_indent and match:
+                findings.extend(self._long_function_finding(chunk, active_name, active_start, chunk.line_start + index - 1))
+                active_start = None
+                active_name = None
+
+            if match:
+                active_start = chunk.line_start + index
+                active_name = match.group(2) if match.lastindex and match.lastindex >= 2 else "function"
+                active_indent = indent
+
+        if active_start is not None:
+            findings.extend(self._long_function_finding(chunk, active_name, active_start, chunk.line_end))
+
+        return findings
+
+    def _check_branch_density(self, chunk: CodeChunk, lines: list[str]) -> list[Finding]:
+        branch_count = sum(1 for line in lines if PYTHON_BRANCH.match(line) or JS_BRANCH.match(line))
+        if branch_count <= MAX_BRANCHES_PER_CHUNK:
+            return []
+
+        return [
+            self._finding(
+                "High branching complexity",
+                Severity.medium,
+                chunk,
+                chunk.line_start,
+                chunk.line_end,
+                f"This code section contains {branch_count} control-flow branches.",
+                "Extract decision-heavy logic into named helpers and add focused tests around each path.",
+            )
+        ]
+
+    def _check_todo_comments(self, chunk: CodeChunk, lines: list[str]) -> list[Finding]:
+        findings: list[Finding] = []
+        for offset, line in enumerate(lines):
+            if TODO_COMMENT.search(line):
+                line_number = chunk.line_start + offset
+                findings.append(
+                    self._finding(
+                        "Unresolved maintenance comment",
+                        Severity.low,
+                        chunk,
+                        line_number,
+                        line_number,
+                        "A TODO/FIXME/HACK marker indicates known unfinished or fragile code.",
+                        "Convert the comment into a tracked issue or resolve it before shipping.",
+                    )
+                )
+        return findings
+
+    def _check_short_names(self, chunk: CodeChunk, lines: list[str]) -> list[Finding]:
+        findings: list[Finding] = []
+        for offset, line in enumerate(lines):
+            match = self._definition_match(line)
+            if not match:
+                continue
+
+            name = match.group(2) if match.lastindex and match.lastindex >= 2 else ""
+            if len(name) < MIN_MEANINGFUL_NAME_LENGTH and name not in {"id"}:
+                line_number = chunk.line_start + offset
+                findings.append(
+                    self._finding(
+                        "Very short symbol name",
+                        Severity.low,
+                        chunk,
+                        line_number,
+                        line_number,
+                        f"The symbol `{name}` is short enough to make intent unclear.",
+                        "Use a descriptive function, class, or variable name that explains the role of this code.",
+                    )
+                )
+        return findings
+
+    def _long_function_finding(
+        self,
+        chunk: CodeChunk,
+        name: str | None,
+        line_start: int,
+        line_end: int,
+    ) -> list[Finding]:
+        if line_end - line_start + 1 <= MAX_FUNCTION_LINES:
+            return []
+
+        return [
+            self._finding(
+                "Long function or class body",
+                Severity.medium,
+                chunk,
+                line_start,
+                line_end,
+                f"`{name or 'This symbol'}` spans more than {MAX_FUNCTION_LINES} lines.",
+                "Extract cohesive helper functions and keep each function centered on one responsibility.",
+            )
+        ]
+
+    def _definition_match(self, line: str) -> re.Match[str] | None:
+        return PYTHON_DEF.match(line) or JS_FUNCTION.match(line)
+
+    def _indent_width(self, line: str) -> int:
+        return len(line) - len(line.lstrip(" "))
+
+    def _finding(
+        self,
+        title: str,
+        severity: Severity,
+        chunk: CodeChunk,
+        line_start: int,
+        line_end: int,
+        description: str,
+        suggested_fix: str,
+    ) -> Finding:
+        return Finding(
+            title=title,
+            severity=severity,
+            file_path=chunk.file_path,
+            line_start=line_start,
+            line_end=line_end,
+            description=description,
+            why_it_matters="Maintainable code is easier to review, test, debug, and safely extend during a hackathon demo.",
+            suggested_fix=suggested_fix,
+            agent_source=self.name,
+        )

tests/test_graph_progress.py

@@ -23,7 +23,9 @@ async def test_run_with_progress_yields_real_stages_and_report(tmp_path: Path):
     assert any("Crawler Agent" in event for event in events if isinstance(event, str))
     assert any("Security Agent" in event for event in events if isinstance(event, str))
     assert any("Performance Agent" in event for event in events if isinstance(event, str))
+    assert any("Quality Agent" in event for event in events if isinstance(event, str))
     assert isinstance(events[-1], AuditReport)
     assert len(events[-1].findings) == 2
     assert "Security Agent" in events[-1].agents_run
     assert "Performance Agent" in events[-1].agents_run
+    assert "Quality Agent" in events[-1].agents_run

tests/test_quality_agent.py

@@ -0,0 +1,57 @@
+import pytest
+
+from app.agents.quality_agent import QualityAgent
+from app.schemas import CodeChunk, Severity
+
+
+@pytest.mark.anyio
+async def test_quality_agent_flags_todo_comments():
+    chunk = CodeChunk(
+        file_path="service.py",
+        language="Python",
+        line_start=12,
+        line_end=12,
+        content="# TODO: handle retry failures",
+    )
+
+    output = await QualityAgent().analyze([chunk])
+
+    assert output.findings[0].title == "Unresolved maintenance comment"
+    assert output.findings[0].severity == Severity.low
+    assert output.findings[0].line_start == 12
+
+
+@pytest.mark.anyio
+async def test_quality_agent_flags_high_branch_density():
+    lines = [f"if condition_{index}:" for index in range(30)]
+    chunk = CodeChunk(
+        file_path="rules.py",
+        language="Python",
+        line_start=1,
+        line_end=len(lines),
+        content="\n".join(lines),
+    )
+
+    output = await QualityAgent().analyze([chunk])
+
+    assert output.findings[0].title == "High branching complexity"
+    assert output.findings[0].severity == Severity.medium
+
+
+@pytest.mark.anyio
+async def test_quality_agent_flags_long_function():
+    lines = ["def process_everything():"]
+    lines.extend(f"    value_{index} = {index}" for index in range(85))
+    chunk = CodeChunk(
+        file_path="processor.py",
+        language="Python",
+        line_start=30,
+        line_end=30 + len(lines) - 1,
+        content="\n".join(lines),
+    )
+
+    output = await QualityAgent().analyze([chunk])
+
+    assert output.findings[0].title == "Long function or class body"
+    assert output.findings[0].severity == Severity.medium
+    assert output.findings[0].line_start == 30