File size: 4,721 Bytes
d939b2e
a3859cd
 
76f9fab
1251fb0
a3859cd
 
d860535
 
3ab2a42
a3859cd
80b64e3
a3859cd
08c6f9e
4d69fe0
a3859cd
 
 
d939b2e
a3859cd
 
 
 
a6031f7
 
d860535
 
 
 
 
 
 
a6031f7
 
 
 
 
a3859cd
 
 
 
 
 
 
 
 
 
08c6f9e
a3859cd
a79668c
 
 
 
a3859cd
 
 
 
a79668c
 
a3859cd
 
a79668c
a3859cd
 
a79668c
a3859cd
a6031f7
 
 
d860535
 
 
 
 
 
a3859cd
a79668c
 
 
a3859cd
 
 
 
 
95e7383
a6031f7
 
 
 
a79668c
5457bc2
 
 
 
 
a3859cd
227aada
2b2abfe
a3859cd
b49f3c0
a3859cd
a79668c
 
 
a3859cd
 
 
a79668c
 
 
a3859cd
 
 
a79668c
 
 
a3859cd
 
 
a79668c
a3859cd
a79668c
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
from app.core.logger import get_logger
from app.tools.static_analyzer import run_full_analysis
from app.tools.pytest_runner import run_tests
from app.tools.file_scanner import get_python_files, sanitize_for_prompt
from app.core.llm import call_llm_for_json
from app.core.prompts import BUG_DETECTION_PROMPT
from app.models.issue import IssueReport, Bug, Warning, Suggestion
from app.tools.ast_parser import scan_repository_security, scan_secrets_regex
from app.tools.dependency_scanner import scan_dependencies


logger = get_logger(__name__)


async def run(local_path: str) -> IssueReport:
    """
    Detect bugs, warnings and suggestions in a repository.
    """
    logger.info("Starting bug detection agent", extra={"path": local_path})

    # Step 1 - Run static analysis
    static_results = run_full_analysis(local_path)

    # Step 1.5 - AST-based security scan (no LLM, zero cost, deterministic)
    ast_security_findings = scan_repository_security(local_path)
    # Regex-based secret scan (covers non-Python files: .env, YAML, JS, etc.)
    secret_findings = scan_secrets_regex(local_path)
    logger.info("Secret scan complete", extra={"findings": len(secret_findings)})

    # Dependency vulnerability scan
    dep_vulnerabilities = scan_dependencies(local_path)
    logger.info("Dependency scan complete", extra={"vulnerabilities": len(dep_vulnerabilities)})
    logger.info(
        "AST security scan complete",
        extra={"findings": len(ast_security_findings)},
    )

    # Step 2 - Run tests
    test_results = run_tests(local_path)

    # Step 3 - Read sample of code for LLM
    python_files = get_python_files(local_path)
    code_samples = []
    for file_path in python_files[:5]:  # Limit to first 5 files
        try:
            with open(file_path, "r", encoding="utf-8", errors="ignore") as f:
                content = f.read()[:2000]  # Limit to 2000 chars per file
                content = sanitize_for_prompt(content)
                code_samples.append(f"### {file_path}\n{content}")
        except Exception as e:  # nosec B110
            logger.warning(
                "Failed to read file", extra={"file": file_path, "error": str(e)}
            )

    # Step 4 - Build context for LLM
    code_info = f"""
Static Analysis Results:
Ruff Issues: {len(static_results["ruff"]["issues"])}
Bandit Issues: {len(static_results["bandit"]["issues"])}

Ruff Output:
{chr(10).join(static_results["ruff"]["issues"][:20])}

Bandit Output:
{chr(10).join(static_results["bandit"]["issues"][:20])}

AST Security Findings:
{chr(10).join(f"[{f['severity'].upper()}] {f['rule']} at line {f['line']} in {f['file']}: {f['detail']}" for f in ast_security_findings[:20])}

Dependency Vulnerabilities:
{chr(10).join(f"[VULN] {v['package']}=={v['version']} {v['id']}: {v['description'][:100]}" for v in dep_vulnerabilities[:10]) or "None found"}

Secret Scan (non-Python files):
{chr(10).join(f"[{f['severity'].upper()}] {f['rule']} at line {f['line']} in {f['file']}" for f in secret_findings[:10]) or "None found"}

Test Results:
Passed: {test_results["passed"]}
Failed: {test_results["failed"]}
Errors: {test_results["errors"]}

Code Samples:
{chr(10).join(code_samples)}
"""

    # Step 5 - Skip LLM if static analysis found nothing and no code to review
    total_static = (
        len(static_results["ruff"]["issues"])
        + len(static_results["bandit"]["issues"])
        + len(ast_security_findings)
    )
    if total_static == 0 and not code_samples:
        logger.info("No static issues and no code samples — skipping LLM call")
        return IssueReport()

    # Call LLM only when there's something meaningful to analyse
    prompt = BUG_DETECTION_PROMPT.format(code_info=code_info)
    llm_data = await call_llm_for_json(prompt, task="bug_detection")
    # call_llm_for_json handles JSON parsing and retry internally; llm_data is already a dict

    # Step 6 - Build IssueReport
    critical = [
        Bug(**bug)
        if isinstance(bug, dict)
        else Bug(description=str(bug), file="unknown")
        for bug in llm_data.get("critical", [])
    ]
    warnings = [
        Warning(**w)
        if isinstance(w, dict)
        else Warning(description=str(w), file="unknown")
        for w in llm_data.get("warnings", [])
    ]
    suggestions = [
        Suggestion(**s)
        if isinstance(s, dict)
        else Suggestion(description=str(s), file="unknown")
        for s in llm_data.get("suggestions", [])
    ]

    report = IssueReport(critical=critical, warnings=warnings, suggestions=suggestions)

    logger.info(
        "Bug detection complete",
        extra={"critical": report.total_critical, "warnings": report.total_warnings},
    )
    return report