Spaces:
Running
Running
| 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 | |