AI Agents

{"path": "app/auth/login.py"}

### app/auth/login.py
```python
from fastapi import APIRouter, Depends
from app.core.database import get_db
...

**Implementation** (from `pr_review_service.py:291-300`):
```python
async def _read_file(self, path: str) -> str:
    content = await github_client.get_file_content(
        self.owner, self.repo, path, self.head_sha
    )
    if not content:
        return f"File not found or empty: {path}"
    if len(content) > 8000:
        content = content[:8000] + "\n# ... (truncated at 8 000 chars)"
    ext = path.rsplit(".", 1)[-1].lower() if "." in path else ""
    return f"### {path}\n```{ext}\n{content}\n```"

{"query": "rate limiting strategy"}

- Uses Redis-backed rate limiter with sliding window (60 req/min per user)
- Rate limit errors return 429 with Retry-After header

async def _search_project_memory(self, query: str) -> str:
    results = await memory_adapter.search_relevant(
        repo=self.repo_full_name, query=query, developer=None, top_k=8
    )
    if not results:
        return "No relevant project memories found."
    lines = [f"- {m.get('memory', m.get('content', ''))}" for m in results]
    return "\n".join(lines)

{"developer": "alice", "query": "error handling habits"}

@alice memory:
- @alice occasionally forgets to handle token expiry in edge cases
- @alice consistently writes comprehensive test coverage

async def _search_developer_memory(self, developer: str, query: str) -> str:
    results = await memory_adapter.search_relevant(
        repo=self.repo_full_name, query=query, developer=developer, top_k=5
    )
    if not results:
        return f"No memories found for @{developer}."
    lines = [f"- {m.get('memory', m.get('content', ''))}" for m in results]
    return f"@{developer} memory:\n" + "\n".join(lines)

{"paths": ["app/auth/token_service.py", "app/auth/login.py"]}

File experts (most commits on these files):
  @alice — 12 PRs
  @bob — 7 PRs
Related past PRs:
  PR #42 [APPROVE] by @alice: Refactor auth service
  PR #38 [REQUEST_CHANGES] by @bob: Fix login redirect

async def _get_file_history(self, paths: list[str]) -> str:
    experts, related = await asyncio.gather(
        graph_builder.get_file_experts(self.repo_full_name, paths[:10], top_k=5),
        graph_builder.get_related_prs(self.repo_full_name, paths[:10], top_k=5),
        return_exceptions=True,
    )
    lines: list[str] = []
    if isinstance(experts, list) and experts:
        lines.append("File experts (most commits on these files):")
        for e in experts:
            lines.append(f"  @{e['login']} — {e['touch_count']} PRs")
    if isinstance(related, list) and related:
        lines.append("Related past PRs:")
        for p in related:
            lines.append(
                f"  PR #{p['number']} [{p.get('verdict', '?')}] "
                f"by @{p.get('author', '?')}: {p.get('title', '')}"
            )
    return "\n".join(lines) if lines else "No history found for these files."

{"numbers": [42, 38]}

Issue #42 [open]: Login redirect breaks on mobile
  Users report that after login, mobile browsers redirect to...
Issue #38 [closed]: Token expiry not handled
  When JWT token expires, API returns 500 instead of 401

async def _get_issue_details(self, numbers: list[int]) -> str:
    results = await asyncio.gather(
        *[github_client.get_issue(self.owner, self.repo, n) for n in numbers[:5]],
        return_exceptions=True,
    )
    lines: list[str] = []
    for n, r in zip(numbers, results):
        if isinstance(r, Exception) or r is None:
            lines.append(f"Issue #{n}: could not fetch")
        else:
            body_preview = (r.get("body") or "")[:200].replace("\n", " ")
            lines.append(
                f"Issue #{n} [{r.get('state', '?')}]: {r.get('title', '')}\n  {body_preview}"
            )
    return "\n".join(lines) if lines else "No issues found."

{"keywords": "login redirect mobile"}

Issue #42: Login redirect breaks on mobile
Issue #39: Mobile Safari login flow broken

async def _search_open_issues(self, keywords: str) -> str:
    kw_set = set(re.findall(r"\b\w{3,}\b", keywords.lower()))
    matches: list[str] = []
    for issue in self.candidate_issues:
        text = f"{issue.get('title') or ''} {issue.get('body') or ''}".lower()
        if len(kw_set & set(re.findall(r"\b\w{3,}\b", text))) >= 2:
            matches.append(f"Issue #{issue['number']}: {issue.get('title', '')}")
    if not matches:
        return "No matching open issues found."
    return "\n".join(matches[:8])

{"query": "rate limiting", "source": "linear"}

Linked linear issues for 'rate limiting':
  #ENG-42 [in progress]: Implement rate limiting for public API
  #ENG-38 [done]: Add Redis cache for rate limit counters

async def _get_linked_issues(self, query: str, source: str = "github") -> str:
    from app.mcp.client import mcp_client
    
    try:
        if source == "linear":
            issues = await mcp_client.get_linear_issues(team_id="", query=query)
        else:
            issues = await mcp_client.get_github_issues(
                repo=self.repo_full_name, query=query
            )
        
        if not issues:
            return f"No {source} issues found for query: {query!r}"
        
        lines = [f"Linked {source} issues for {query!r}:"]
        for issue in issues[:10]:
            number = issue.get("number") or issue.get("id", "?")
            title = issue.get("title", "(no title)")
            state = issue.get("state", "")
            state_tag = f" [{state}]" if state else ""
            body = (issue.get("description") or issue.get("body") or "")[:120].replace("\n", " ")
            desc = f" — {body}" if body else ""
            lines.append(f"  #{number}{state_tag}: {title}{desc}")
        return "\n".join(lines)
    except Exception as exc:
        logger.warning("_get_linked_issues failed: %s", exc)
        return f"Could not fetch {source} issues: {exc}"

{"files": ["app/auth/token_service.py"]}

Related Sentry errors for modified files:
  [42x] JWTDecodeError: Invalid token signature — culprit: app.auth.token_service.decode_token (last seen: 2024-03-10)
  [12x] KeyError: 'exp' in JWT payload — culprit: app.auth.token_service.validate_expiry (last seen: 2024-03-09)

async def _get_related_errors(self, files: list[str]) -> str:
    from app.mcp.client import mcp_client
    from app.core.config import settings
    
    if not settings.SENTRY_MCP_URL:
        return (
            "Sentry integration not configured "
            "(set SENTRY_MCP_URL + SENTRY_AUTH_TOKEN to enable)"
        )
    
    all_errors: list[dict] = []
    for path in files[:5]:
        try:
            errors = await mcp_client.get_sentry_errors(
                project=self.repo,
                filename=path,
            )
            all_errors.extend(errors)
        except Exception as exc:
            logger.warning("_get_related_errors failed for %s: %s", path, exc)
    
    if not all_errors:
        return "No related Sentry errors found for the modified files."
    
    seen: set[str] = set()
    lines = ["Related Sentry errors for modified files:"]
    for err in all_errors[:10]:
        key = str(err.get("id") or err.get("title", ""))
        if key in seen:
            continue
        seen.add(key)
        title = err.get("title", "(unknown error)")
        culprit = err.get("culprit", "")
        count = err.get("count", "?")
        last_seen = err.get("last_seen", "")
        lines.append(
            f"  [{count}x] {title}"
            + (f" — culprit: {culprit}" if culprit else "")
            + (f" (last seen: {last_seen})" if last_seen else "")
        )
    return "\n".join(lines)

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."""

[
  {
    "severity": "HIGH",
    "file": "app/auth/token_service.py",
    "line": 42,
    "issue": "JWT tokens validated without signature verification — attacker can forge tokens",
    "fix": "Use jwt.decode(..., verify_signature=True) and pass secret key"
  }
]

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."""

[
  {
    "impact": "HIGH",
    "file": "app/services/user_service.py",
    "line": 78,
    "issue": "N+1 query: loops over users and fetches posts individually — 100 users = 101 queries",
    "fix": "Use selectinload(User.posts) in initial query to eager-load"
  }
]

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."""

[
  {
    "severity": "MEDIUM",
    "file": "app/services/payment_service.py",
    "line": 123,
    "issue": "No test coverage for stripe_webhook failure path",
    "fix": "Add test case mocking Stripe API error to ensure graceful degradation"
  }
]

synthesis_prompt = f"""You are the final synthesizer for a parallel code review system.
Three specialized agents have analyzed PR #{pr.get('number')} — "{pr_title}" by @{author}.

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
- verdict = "comment" if borderline (medium issues only)
- Deduplicate if multiple agents flagged the same issue
- inline_comments: max 8, most impactful only
- Be constructive and specific"""