Spaces:
Sleeping
Sleeping
| """ | |
| Shared analysis pipeline, used by both the synchronous /analyze route | |
| and the async Celery task triggered by /analyze/async. | |
| """ | |
| import asyncio | |
| import time | |
| from app.agents import ( | |
| bug_detection_agent, | |
| code_review_agent, | |
| report_generator_agent, | |
| repo_analysis_agent, | |
| specialist_review_agent, | |
| test_generation_agent, | |
| ) | |
| from app.core.cache import get_cached_report, set_cached_report | |
| from app.core.logger import get_logger | |
| from app.core.metrics import metrics | |
| from app.models.issue import IssueReport | |
| from app.models.report import EngineeringReport | |
| from app.models.review import GeneratedTests, ReviewSuggestions | |
| from app.tools.github_tool import ( | |
| clone_repository, | |
| delete_repository, | |
| resolve_remote_head_sha, | |
| ) | |
| from app.models.report import SecurityScanResult | |
| 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_full_analysis( | |
| github_url: str, base_sha: str | None = None | |
| ) -> EngineeringReport: | |
| """ | |
| Full multi-agent pipeline with a Redis-backed result cache. | |
| If the remote repo's current HEAD SHA can be resolved and a cached | |
| report exists for (github_url, head_sha), that report is returned | |
| immediately — no clone, no agent runs. Otherwise the pipeline runs | |
| normally and the result is cached for subsequent identical requests. | |
| """ | |
| head_sha = resolve_remote_head_sha(github_url) | |
| if head_sha: | |
| cached = get_cached_report(github_url, head_sha) | |
| if cached is not None: | |
| metrics.record_cache_event(hit=True) | |
| logger.info( | |
| "Cache hit — returning cached analysis", | |
| extra={"url": github_url, "head_sha": head_sha}, | |
| ) | |
| cached.from_cache = True | |
| return cached | |
| metrics.record_cache_event(hit=False) | |
| report = await _run_pipeline(github_url) | |
| if head_sha: | |
| report.repository.head_sha = head_sha | |
| set_cached_report(github_url, head_sha, report) | |
| return report | |
| async def _run_pipeline(github_url: str) -> EngineeringReport: | |
| """The uncached clone-and-analyze pipeline.""" | |
| local_path = None | |
| _start = time.monotonic() | |
| try: | |
| local_path = clone_repository(github_url) | |
| _t = time.monotonic() | |
| metadata = await repo_analysis_agent.run(github_url, local_path) | |
| metrics.record_run("repo_analysis_agent", time.monotonic() - _t, success=True) | |
| _t = time.monotonic() | |
| results = await asyncio.gather( | |
| bug_detection_agent.run(local_path), | |
| test_generation_agent.run(local_path), | |
| code_review_agent.run(local_path, metadata), | |
| specialist_review_agent.run(local_path, metadata, IssueReport()), | |
| return_exceptions=True, | |
| ) | |
| any_failed = any(isinstance(r, BaseException) for r in results) | |
| metrics.record_run( | |
| "parallel_agents", time.monotonic() - _t, success=not any_failed | |
| ) | |
| issues = results[0] if not isinstance(results[0], BaseException) else IssueReport() | |
| tests = results[1] if not isinstance(results[1], BaseException) else GeneratedTests() | |
| review = ( | |
| results[2] | |
| if not isinstance(results[2], BaseException) | |
| else ReviewSuggestions(summary="Review failed.", overall_score=0.0) | |
| ) | |
| specialist_review = results[3] if not isinstance(results[3], BaseException) else None | |
| for i, r in enumerate(results): | |
| if isinstance(r, BaseException): | |
| logger.error("Agent failed", extra={"agent_index": i, "error": str(r)}) | |
| _t = time.monotonic() | |
| report = await report_generator_agent.run( | |
| metadata, issues, tests, review, specialist_review | |
| ) | |
| # Already computed inside bug_detection_agent, but we surface them on the report too | |
| # These are fast (no LLM) so re-running is acceptable | |
| ast_findings = scan_repository_security(local_path) | |
| secret_findings = scan_secrets_regex(local_path) | |
| dep_vulns = scan_dependencies(local_path) | |
| high = sum(1 for f in ast_findings + secret_findings if f.get("severity") == "high") | |
| critical = sum(1 for f in ast_findings if f.get("severity") == "critical") | |
| report.security_scan = SecurityScanResult( | |
| ast_findings=ast_findings, | |
| secret_findings=secret_findings, | |
| dependency_vulnerabilities=dep_vulns, | |
| total_high=high, | |
| total_critical=critical, | |
| ) | |
| metrics.record_run( | |
| "report_generator_agent", time.monotonic() - _t, success=True | |
| ) | |
| metrics.record_run("full_pipeline", time.monotonic() - _start, success=True) | |
| return report | |
| finally: | |
| if local_path: | |
| delete_repository(local_path) |