Parallel Review Agents

# .env
PARALLEL_REVIEW_AGENTS=true

use_parallel = getattr(settings, 'PARALLEL_REVIEW_AGENTS', False)
if use_parallel:
    logger.info("Starting parallel AI analysis (3 specialized agents concurrently)...")
    review_result = await ai_service.analyze_pull_request_parallel(
        pr, diff, files, tool_executor, issue_refs=issue_refs
    )
else:
    logger.info("Starting agentic AI analysis (Claude fetches context on demand)...")
    review_result = await ai_service.analyze_pull_request_agentic(
        pr, diff, files, tool_executor, issue_refs=issue_refs
    )

SECURITY_AGENT_PROMPT = """You are a specialized security code reviewer. 
Focus EXCLUSIVELY on security issues:
- Injection vulnerabilities (SQL, command, path traversal, SSRF)
- Authentication and authorization flaws
- Secrets/credentials accidentally committed
- Insecure dependencies or imports
- Input validation gaps at trust boundaries
- Cryptographic weaknesses
- Sensitive data exposure (PII in logs, unencrypted storage)

For each issue found: severity (CRITICAL/HIGH/MEDIUM/LOW), file:line, what the risk is, concrete fix.
If no security issues: say "No security issues found" — do NOT invent issues.
Be terse. Output JSON-serializable structured findings."""

SECURITY_TOOLS = [t for t in REVIEW_TOOLS if t["name"] in {
    "read_file", "search_project_memory", "get_issue_details", "search_open_issues"
}]

PERFORMANCE_AGENT_PROMPT = """You are a specialized performance code reviewer.
Focus EXCLUSIVELY on performance issues:
- N+1 database queries (loop + individual queries)
- Missing indexes or inefficient query patterns
- Unbounded loops or O(n²)+ algorithms where O(n log n) is feasible
- Memory leaks (unclosed resources, unbounded caches, circular refs)
- Blocking I/O in async contexts
- Unnecessary serialization/deserialization in hot paths
- Large payload transfers that could be paginated/streamed

For each issue found: impact (HIGH/MEDIUM/LOW), file:line, what the bottleneck is, concrete fix.
If no performance issues: say "No performance issues found" — do NOT invent issues.
Be terse. Output JSON-serializable structured findings."""

PERFORMANCE_TOOLS = [t for t in REVIEW_TOOLS if t["name"] in {
    "read_file", "get_file_history", "search_project_memory"
}]

STYLE_AGENT_PROMPT = """You are a specialized code quality reviewer.
Focus EXCLUSIVELY on code quality, tests, and maintainability:
- Missing or inadequate test coverage for new logic
- Functions/methods that are too complex (>20 lines, deep nesting)
- Unclear variable/function naming that hinders readability
- Missing error handling for operations that can fail
- Dead code, unused imports, or leftover debug statements
- API contract breakages (changed signatures, removed fields)
- Missing or outdated docstrings on public interfaces

For each issue: severity (HIGH/MEDIUM/LOW), file:line, what the issue is, concrete fix.
If no style/quality issues: say "No style issues found" — do NOT invent issues.
Be terse. Output JSON-serializable structured findings."""

STYLE_TOOLS = [t for t in REVIEW_TOOLS if t["name"] in {
    "read_file", "search_developer_memory", "search_project_memory", "search_open_issues"
}]

async def analyze_pull_request_parallel(
    self, pr: dict, diff: str, files: list[dict], tool_executor, issue_refs: list[int] | None = None
) -> dict:
    # Build shared context (same for all 3 agents)
    context_block = f"""PR #{pr.get('number')} — {pr_title}
Author: @{author}
Files changed ({len(changed_files)}): {', '.join(changed_files[:15])}
Issue refs: {issue_refs or []}

--- DIFF ---
{diff[:12000]}
"""

    # Run all 3 agents concurrently
    security_task = self._run_specialized_agent("security", SECURITY_AGENT_PROMPT, context_block, SECURITY_TOOLS, tool_executor)
    performance_task = self._run_specialized_agent("performance", PERFORMANCE_AGENT_PROMPT, context_block, PERFORMANCE_TOOLS, tool_executor)
    style_task = self._run_specialized_agent("style", STYLE_AGENT_PROMPT, context_block, STYLE_TOOLS, tool_executor)

    security_out, performance_out, style_out = await asyncio.gather(
        security_task, performance_task, style_task, return_exceptions=True
    )

    # Synthesis agent combines all findings
    return await self._synthesize_review(pr, diff, files, security_out, performance_out, style_out, issue_refs)

async def _run_specialized_agent(
    self, agent_name: str, system_prompt: str, context: str, tools: list[dict], tool_executor, max_rounds: int = 5
) -> str:
    messages = [{"role": "user", "content": context}]

    for round_num in range(max_rounds):
        response = await self.client.messages.create(
            model=self.model,
            max_tokens=2048,
            system=system_prompt,
            tools=tools,
            messages=messages,
        )

        if response.stop_reason == "end_turn":
            return "".join(b.text for b in response.content if hasattr(b, "text"))

        if response.stop_reason == "tool_use":
            # Execute tools → append results → continue
            for block in response.content:
                if block.type == "tool_use":
                    result = await tool_executor.execute(block.name, block.input)
                    tool_results.append({"type": "tool_result", "tool_use_id": block.id, "content": str(result)[:4000]})
            messages.append({"role": "assistant", "content": response.content})
            messages.append({"role": "user", "content": tool_results})

    return "[no output]"

async def _synthesize_review(
    self, pr: dict, diff: str, files: list[dict],
    security_findings: str, performance_findings: str, style_findings: str, issue_refs: list[int] | None
) -> dict:
    synthesis_prompt = f"""You are the final synthesizer for a parallel code review system.
Three specialized agents have analyzed PR #{pr.get('number')}.

SECURITY AGENT FINDINGS:
{security_findings}

PERFORMANCE AGENT FINDINGS:
{performance_findings}

STYLE/QUALITY AGENT FINDINGS:
{style_findings}

Based on ALL findings above, produce a final unified code review. Respond in this EXACT JSON format:
{{
  "verdict": "approved" | "changes_requested" | "comment",
  "summary": "2-3 sentence overall assessment",
  "security_issues": [],
  "performance_issues": [],
  "style_issues": [],
  "inline_comments": [{{"path": "file/path.py", "line": 42, "body": "specific comment"}}],
  "memory_insights": "patterns worth remembering about this author/codebase"
}}

Rules:
- verdict = "changes_requested" if ANY critical/high security or performance issue
- verdict = "approved" if only low/medium style issues or no issues
- Deduplicate if multiple agents flagged the same issue
- inline_comments: max 8, most impactful only
"""

    response = await self.client.messages.create(
        model=self.model, max_tokens=3000,
        system="You are a senior engineer synthesizing a parallel code review. Output valid JSON only.",
        messages=[{"role": "user", "content": synthesis_prompt}],
    )

    return json.loads(response.content[0].text)

def safe_result(result, name: str) -> str:
    if isinstance(result, Exception):
        logger.error(f"{name} agent failed: {result}")
        return f"[{name} agent error: {result}]"
    return result

security_findings = safe_result(security_out, "security")
performance_findings = safe_result(performance_out, "performance")
style_findings = safe_result(style_out, "style")

# Enable detailed logging
export LOG_LEVEL=DEBUG

# Look for agent execution logs
tail -f app.log | grep "\[security\]\|\[performance\]\|\[style\]"

INFO: Starting parallel AI analysis (3 specialized agents concurrently)...
INFO: [security] round 1: stop_reason=tool_use
INFO: [security] tool call: read_file({'path': 'app/auth/login.py'})
INFO: [performance] round 1: stop_reason=tool_use
INFO: [performance] tool call: get_file_history({'paths': ['app/db/queries.py']})
INFO: [style] round 1: stop_reason=end_turn
INFO: [security] round 2: stop_reason=end_turn
INFO: [performance] round 2: stop_reason=end_turn