salim0986/graph-bug-ai
0
1"""2LangGraph Workflow for AI Code Review - Phase 4.13Orchestrates the entire review process using LangGraph StateGraph4"""5 6from typing import TypedDict, List, Dict, Any, Annotated, Optional, Literal7from typing_extensions import NotRequired8from langgraph.graph import StateGraph, START, END9from langgraph.graph.message import add_messages10from dataclasses import dataclass, field, asdict11from datetime import datetime12from .logger import setup_logger13from .context_builder import ContextBuilder14from .gemini_client import GeminiClient, create_gemini_client15from .analyzer import FileChange16from .temporary_graph import TemporaryGraphBuilder, TemporaryVectorBuilder17from .context_merger import ContextMerger18from .parser import UniversalParser, detect_language_from_filename19from .review_validator import ReviewValidator20from .review_formatter import ReviewFormatter21from .review_schema import ReviewOutput, CritiqueOutput, CritiqueFinding22from sentence_transformers import SentenceTransformer23 24logger = setup_logger(__name__)25 26 27# ============================================================================28# STATE DEFINITION29# ============================================================================30 31class ReviewState(TypedDict):32 """33 Complete state for the code review workflow34 35 This state is passed through all nodes and maintains the entire36 review context and progress.37 """38 # PR Metadata39 pr_number: int40 repo_id: str41 pr_title: str42 pr_description: Optional[str]43 base_ref: NotRequired[str] # Base branch (e.g., "main")44 head_ref: NotRequired[str] # Head branch (e.g., "feature/new-feature")45 46 # Input Data47 files: List[Dict[str, Any]] # Raw file changes from GitHub48 github_client: NotRequired[Any] # Client for fetching full file contents49 installation_id: NotRequired[int] # GitHub app installation ID50 51 # Context (from Phase 3)52 pr_context: NotRequired[Dict[str, Any]] # PRContext from context_builder53 _merged_contexts: NotRequired[Dict[str, Any]] # Merged GraphRAG contexts (temporary + permanent)54 55 # Review Strategy56 review_strategy: NotRequired[Literal["quick", "standard", "deep"]]57 requires_deep_review: NotRequired[bool]58 risk_level: NotRequired[str]59 60 # File Prioritization61 high_priority_files: NotRequired[List[str]]62 medium_priority_files: NotRequired[List[str]]63 low_priority_files: NotRequired[List[str]]64 65 # Review Generation66 file_reviews: NotRequired[Dict[str, Any]]67 review_summary: NotRequired[Dict[str, Any]]68 overall_summary: NotRequired[str]69 70 # M6: structured review output (ReviewOutput.model_dump())71 structured_review: NotRequired[Dict[str, Any]]72 # M8: critique + revised review from reflect/revise nodes73 critique: NotRequired[Dict[str, Any]]74 revised_review: NotRequired[Dict[str, Any]]75 # M8: was LLM-assisted routing used for this PR?76 llm_route_used: NotRequired[bool]77 78 # Execution metrics79 parse_failure_total: NotRequired[int] # gemini-2.5-flash-lite, flash, or pro80 81 # Model Selection82 selected_model: NotRequired[str] # gemini-2.5-flash-lite, flash, or pro83 84 # Progress Tracking85 status: NotRequired[str] # queued, analyzing, reviewing, completed, failed86 current_step: NotRequired[str]87 processed_files: NotRequired[int]88 total_files: NotRequired[int]89 90 # Error Handling91 errors: NotRequired[List[Dict[str, Any]]]92 retry_count: NotRequired[int]93 94 # Timing95 started_at: NotRequired[str]96 completed_at: NotRequired[str]97 98 # Messages (for LangGraph message passing)99 messages: Annotated[List[Dict[str, Any]], add_messages]100 101 102# ============================================================================103# WORKFLOW CONFIGURATION104# ============================================================================105 106@dataclass107class WorkflowConfig:108 """Configuration for the review workflow"""109 110 # Model selection thresholds111 quick_review_max_files: int = 3112 quick_review_max_additions: int = 100113 114 standard_review_max_files: int = 10115 standard_review_max_additions: int = 500116 117 # Gemini model configuration118 flash_lite_model: str = "gemini-2.5-flash-lite"119 flash_model: str = "gemini-2.5-flash"120 pro_model: str = "gemini-2.5-pro"121 122 # Rate limiting123 max_requests_per_minute: int = 60124 max_tokens_per_request: int = 30000125 126 # Retry configuration127 max_retries: int = 3128 retry_delay_seconds: int = 2129 130 # Review configuration131 max_files_per_batch: int = 5132 context_window_tokens: int = 25000133 134 # Priority thresholds135 high_priority_complexity: int = 70136 high_priority_issues: int = 3 # critical + high issues137 138 139# ============================================================================140# WORKFLOW BUILDER141# ============================================================================142 143class CodeReviewWorkflow:144 """145 LangGraph workflow for orchestrating AI code reviews146 147 Workflow Steps:148 1. START -> analyze (build context)149 2. analyze -> route (determine review strategy)150 3. route -> review_quick/review_standard/review_deep151 4. review_* -> aggregate152 5. aggregate -> END153 """154 155 def __init__(156 self,157 config: Optional[WorkflowConfig] = None,158 context_builder: Optional[ContextBuilder] = None,159 gemini_client: Optional[GeminiClient] = None160 ):161 self.config = config or WorkflowConfig()162 self.context_builder = context_builder163 self.gemini_client = gemini_client or create_gemini_client()164 self.review_validator = ReviewValidator()165 self.graph = self._build_graph()166 self.compiled = None167 168 # Initialize for temporary GraphRAG (lazy loading)169 self._parser = None170 self._embed_model = None171 172 def _build_graph(self) -> StateGraph:173 """Build the LangGraph StateGraph"""174 175 # Create graph with ReviewState176 workflow = StateGraph(ReviewState)177 178 # Add nodes (we'll implement these in subsequent phases)179 workflow.add_node("analyze", self._analyze_node)180 workflow.add_node("route", self._route_node)181 workflow.add_node("review_quick", self._review_quick_node)182 workflow.add_node("review_standard", self._review_standard_node)183 workflow.add_node("review_deep", self._review_deep_node)184 # M8: reflect + revise nodes inserted between review and aggregate185 workflow.add_node("reflect", self._reflect_node)186 workflow.add_node("revise", self._revise_node)187 workflow.add_node("aggregate", self._aggregate_node)188 workflow.add_node("handle_error", self._error_handler_node)189 190 # Define edges191 workflow.add_edge(START, "analyze")192 workflow.add_edge("analyze", "route")193 194 # Conditional edges from route node195 workflow.add_conditional_edges(196 "route",197 self._route_decision,198 {199 "quick": "review_quick",200 "standard": "review_standard",201 "deep": "review_deep",202 "error": "handle_error"203 }204 )205 206 # M8: all review nodes → reflect → revise → aggregate207 workflow.add_edge("review_quick", "reflect")208 workflow.add_edge("review_standard", "reflect")209 workflow.add_edge("review_deep", "reflect")210 workflow.add_edge("reflect", "revise")211 workflow.add_edge("revise", "aggregate")212 213 # Aggregate goes to END214 workflow.add_edge("aggregate", END)215 216 # Error handler can retry or end217 workflow.add_conditional_edges(218 "handle_error",219 self._error_decision,220 {221 "retry": "analyze",222 "end": END223 }224 )225 226 return workflow227 228 def compile(self):229 """Compile the workflow for execution"""230 self.compiled = self.graph.compile()231 logger.info("Code review workflow compiled successfully")232 return self.compiled233 234 def _get_parser(self) -> UniversalParser:235 """Lazy load UniversalParser for temporary GraphRAG"""236 if self._parser is None:237 self._parser = UniversalParser()238 return self._parser239 240 def _get_embed_model(self) -> SentenceTransformer:241 """Lazy load embedding model for temporary vectors"""242 if self._embed_model is None:243 logger.info("[TempGraphRAG] Loading sentence-transformers model...")244 self._embed_model = SentenceTransformer('all-MiniLM-L6-v2')245 return self._embed_model246 247 async def _build_temporary_graphrag(248 self,249 pr_files: List[Dict[str, Any]],250 repo_full_name: str,251 head_sha: str,252 github_client: Optional[Any] = None,253 installation_id: Optional[int] = None254 ) -> tuple[Optional[TemporaryGraphBuilder], Optional[TemporaryVectorBuilder], int]:255 """256 Build temporary in-memory GraphRAG for PR files257 258 Args:259 pr_files: List of file dicts with filename, status, patch, etc.260 repo_full_name: Full repository name261 head_sha: The commit SHA for the PR head262 github_client: Optional GitHub client to fetch full file content263 installation_id: Installation ID for fetching file contents264 265 Returns:266 Tuple of (temp_graph, temp_vector, parse_failures) or (None, None, 0) on error267 """268 parse_failures = 0269 try:270 logger.info(f"[TempGraphRAG] Building temporary GraphRAG for {len(pr_files)} files")271 272 # Initialize builders273 parser = self._get_parser()274 temp_graph = TemporaryGraphBuilder(parser)275 temp_vector = TemporaryVectorBuilder(self._get_embed_model())276 277 # Process each file278 processed_count = 0279 for file_dict in pr_files:280 try:281 filename = file_dict.get("filename", "")282 language = file_dict.get("language")283 patch = file_dict.get("patch", "")284 status = file_dict.get("status", "modified")285 286 # Skip deleted files287 if status == "deleted":288 continue289 290 # Detect language from filename if not provided by GitHub291 if not language:292 language = detect_language_from_filename(filename)293 if language:294 logger.info(f"[TempGraphRAG] Detected language '{language}' for {filename}")295 296 # Skip files with unsupported languages297 if not language or language == "text":298 logger.warning(f"[TempGraphRAG] Unsupported language for {filename}, skipping")299 continue300 301 # For new/modified files, fetch full content from GitHub API302 content = ""303 304 if github_client and installation_id and head_sha:305 try:306 # Use await since get_file_content is an async method307 content = await github_client.get_file_content(308 repo_full_name=repo_full_name,309 path=filename,310 ref=head_sha,311 installation_id=installation_id312 )313 except Exception as e:314 logger.warning(315 f"[TempGraphRAG] Failed to fetch full content for {filename}: {e}",316 extra={"repo": repo_full_name, "path": filename, "sha": head_sha, "reason": str(e)}317 )318 319 # Fallback to diff patch if fetching failed or client unavailable320 if not content and patch:321 logger.warning(f"[TempGraphRAG] Falling back to patch extraction for {filename}")322 code_lines = []323 for line in patch.split('\n'):324 if line.startswith('+') and not line.startswith('+++'):325 code_lines.append(line[1:]) # Remove + prefix326 elif line.startswith(' '):327 code_lines.append(line[1:]) # Context line328 content = '\n'.join(code_lines)329 330 if not content.strip():331 logger.warning(f"[TempGraphRAG] No content for {filename}, skipping")332 parse_failures += 1333 continue334 335 # Process file with tree-sitter336 file_node = temp_graph.process_file(filename, content, language)337 338 if not file_node or not file_node.nodes:339 parse_failures += 1340 341 # Add nodes to vector index342 if file_node.nodes:343 temp_vector.add_nodes(file_node.nodes)344 processed_count += 1345 346 except Exception as e:347 logger.error(f"[TempGraphRAG] Error processing {filename}: {e}")348 continue349 350 # Build file dependencies351 temp_graph.build_dependencies()352 353 logger.info(354 f"[TempGraphRAG] Built temporary GraphRAG: "355 f"{processed_count} files, {len(temp_graph.nodes)} nodes, "356 f"{len(temp_vector.vectors)} vectors, {parse_failures} failures"357 )358 359 return temp_graph, temp_vector, parse_failures360 361 except Exception as e:362 logger.error(f"[TempGraphRAG] Error building temporary GraphRAG: {e}", exc_info=True)363 return None, None, 0364 365 # ========================================================================366 # NODE IMPLEMENTATIONS (Placeholders for Phase 4.1)367 # ========================================================================368 369 async def _analyze_node(self, state: ReviewState) -> ReviewState:370 """371 Analysis node: Build comprehensive PR context using ContextBuilder372 373 This node:374 1. Converts file dicts to FileChange objects375 2. **NEW**: Builds temporary in-memory GraphRAG for PR files376 3. Builds complete PR context with graph + vector + analysis377 4. **NEW**: Merges temporary and permanent GraphRAG contexts378 5. Extracts risk level and metrics379 6. Prepares state for routing380 """381 logger.info(f"[ANALYZE] Starting analysis for PR #{state['pr_number']}")382 383 state["status"] = "analyzing"384 state["current_step"] = "context_building"385 state["started_at"] = datetime.utcnow().isoformat()386 387 try:388 if not self.context_builder:389 logger.error("ContextBuilder not initialized")390 state["errors"] = [{"message": "ContextBuilder not available", "step": "analyze"}]391 return state392 393 # ==================================================================394 # STEP 1: Build Temporary GraphRAG for PR Files395 # ==================================================================396 logger.info("[ANALYZE] Step 1: Building temporary in-memory GraphRAG")397 temp_graph, temp_vector, parse_failures = await self._build_temporary_graphrag(398 pr_files=state["files"],399 repo_full_name=state["repo_id"],400 head_sha=state.get("head_ref", ""),401 github_client=state.get("github_client"),402 installation_id=state.get("installation_id")403 )404 state["parse_failure_total"] = parse_failures405 406 # Create context merger407 context_merger = ContextMerger(temp_graph=temp_graph, temp_vector=temp_vector)408 409 # Log statistics410 stats = context_merger.get_statistics()411 logger.info(f"[ANALYZE] Temporary GraphRAG stats: {stats}")412 413 # ==================================================================414 # STEP 2: Convert File Dicts to FileChange Objects415 # ==================================================================416 file_changes = []417 for f in state["files"]:418 file_change = FileChange(419 filename=f.get("filename", ""),420 status=f.get("status", "modified"),421 additions=f.get("additions", 0),422 deletions=f.get("deletions", 0),423 patch=f.get("patch"),424 language=f.get("language")425 )426 file_changes.append(file_change)427 428 # ==================================================================429 # STEP 3: Build Comprehensive PR Context (Permanent + Temporary)430 # ==================================================================431 logger.info(f"[ANALYZE] Step 2: Building context for {len(file_changes)} files")432 433 # Extract base_ref and head_ref from state434 base_ref = state.get("base_ref", "main")435 head_ref = state.get("head_ref", "unknown")436 437 # Build PR context with permanent databases438 pr_context = await self.context_builder.build_pr_context(439 pr_number=state["pr_number"],440 repo_id=state["repo_id"],441 title=state["pr_title"],442 description=state.get("pr_description"),443 files=file_changes,444 base_ref=base_ref,445 head_ref=head_ref446 )447 448 # ==================================================================449 # STEP 4: Search Temporary Vectors for Similar Code450 # ==================================================================451 logger.info("[ANALYZE] Step 3: Searching temporary vectors for similar code")452 453 # For each file in temp graph, find similar code within the PR454 temp_similar_results = {}455 if temp_vector and len(temp_vector.vectors) > 0:456 for filename in temp_graph.files.keys():457 file_node = temp_graph.files[filename]458 file_similar = []459 460 # For each node in this file, find similar nodes461 for node in file_node.nodes:462 similar = temp_vector.find_similar_to_node(463 node.id,464 limit=3,465 min_score=0.5 # Lower threshold for PR-internal similarities466 )467 if similar:468 file_similar.extend(similar)469 470 if file_similar:471 temp_similar_results[filename] = file_similar472 logger.info(f"[ANALYZE] Found {len(file_similar)} similar code snippets in {filename} from temporary GraphRAG")473 474 # ==================================================================475 # STEP 5: Merge Temporary + Permanent Contexts476 # ==================================================================477 logger.info("[ANALYZE] Step 4: Merging temporary and permanent contexts")478 479 # Enhance file contexts with merged data480 for file_ctx in pr_context.files:481 # Get permanent context for this file482 permanent_context = {483 "dependencies": [e.model_dump() for e in file_ctx.entities if hasattr(e, 'dependencies')],484 "dependents": [],485 "similar_code": [], # Will be populated from entities486 "imports": getattr(file_ctx, 'imports', []),487 "file_dependencies": getattr(file_ctx, 'file_dependencies', [])488 }489 490 # Extract similar_code from entities491 for entity in file_ctx.entities:492 if hasattr(entity, 'similar_code') and entity.similar_code:493 permanent_context["similar_code"].extend(entity.similar_code)494 495 # Add temporary similar code results496 if file_ctx.filename in temp_similar_results:497 permanent_context["similar_code"].extend(temp_similar_results[file_ctx.filename])498 499 # Merge with temporary context500 merged = context_merger.merge_file_context(501 filename=file_ctx.filename,502 permanent_context=permanent_context503 )504 505 # Store merged context in separate dict (can't add to Pydantic model)506 if "_merged_contexts" not in state:507 state["_merged_contexts"] = {}508 # Convert dataclass to dict for JSON serialization509 state["_merged_contexts"][file_ctx.filename] = asdict(merged)510 511 # Check if GraphRAG data is available (permanent or temporary)512 merged_contexts = state.get("_merged_contexts", {})513 has_graphrag_data = any(514 len(f.entities) > 0 or len(f.dependencies) > 0 or 515 (f.filename in merged_contexts and merged_contexts[f.filename].get("temp_nodes_count", 0) > 0)516 for f in pr_context.files517 )518 519 if not has_graphrag_data:520 logger.warning(521 f"⚠️ No GraphRAG data found for {state['repo_id']}. "522 "Repository may not have been ingested yet. "523 "Similar code and dependency analysis will be limited. "524 f"Run POST /ingest with repo_url to enable full GraphRAG features."525 )526 elif temp_graph and len(temp_graph.files) > 0:527 logger.info(528 f"✅ Using hybrid GraphRAG context: "529 f"{len(temp_graph.files)} files with temporary analysis + permanent database"530 )531 532 # Store context in state (convert Pydantic model to dict)533 state["pr_context"] = pr_context.model_dump()534 535 # Store context merger for use in review nodes536 state["_context_merger"] = context_merger # Internal, not serialized537 538 # Extract key metrics for routing539 state["risk_level"] = pr_context.risk_level540 state["requires_deep_review"] = pr_context.requires_deep_review541 542 # Prioritize files based on complexity and issues543 high_priority = []544 medium_priority = []545 low_priority = []546 547 for file_ctx in pr_context.files:548 if file_ctx.complexity_score >= self.config.high_priority_complexity:549 high_priority.append(file_ctx.filename)550 elif file_ctx.issues_summary.get("critical", 0) + file_ctx.issues_summary.get("high", 0) >= self.config.high_priority_issues:551 high_priority.append(file_ctx.filename)552 elif file_ctx.complexity_score >= 40:553 medium_priority.append(file_ctx.filename)554 else:555 low_priority.append(file_ctx.filename)556 557 state["high_priority_files"] = high_priority558 state["medium_priority_files"] = medium_priority559 state["low_priority_files"] = low_priority560 561 logger.info(f"Analysis complete: risk={pr_context.risk_level}, "562 f"high_priority={len(high_priority)}, "563 f"critical_issues={len(pr_context.critical_issues)}")564 565 state["messages"].append({566 "role": "system",567 "content": f"Context built: {pr_context.total_files} files analyzed, "568 f"{len(pr_context.critical_issues)} critical issues found"569 })570 571 except Exception as e:572 logger.error(f"Error in analyze node: {e}", exc_info=True)573 state["errors"] = state.get("errors", []) + [574 {"message": str(e), "step": "analyze", "timestamp": datetime.utcnow().isoformat()}575 ]576 577 return state578 579 async def _route_node(self, state: ReviewState) -> ReviewState:580 """581 Routing node: Determine review strategy and select Gemini model582 583 Strategy Selection:584 - quick: Small, low-risk PRs (< 3 files, < 100 additions)585 - standard: Typical PRs (< 10 files, < 500 additions)586 - deep: Large, complex, or high-risk PRs587 """588 logger.info(f"[ROUTE] Determining review strategy for PR #{state['pr_number']}")589 590 state["current_step"] = "routing"591 592 try:593 pr_context = state.get("pr_context", {})594 total_files = pr_context.get("total_files", len(state["files"]))595 total_additions = pr_context.get("total_additions", 0)596 risk_level = state.get("risk_level", "low")597 requires_deep = state.get("requires_deep_review", False)598 599 # Determine strategy600 if total_files <= self.config.quick_review_max_files and \601 total_additions <= self.config.quick_review_max_additions and \602 risk_level == "low" and \603 not requires_deep:604 strategy = "quick"605 606 elif total_files > self.config.standard_review_max_files or \607 total_additions > self.config.standard_review_max_additions or \608 risk_level in ["high", "critical"] or \609 requires_deep:610 strategy = "deep"611 612 else:613 # M8: borderline "standard" PRs get an LLM-assisted routing decision614 strategy = await self._llm_route_standard(state)615 state["llm_route_used"] = True616 617 state["review_strategy"] = strategy618 619 # Select Gemini model based on strategy and PR characteristics620 selected_model = self.gemini_client.select_model(621 total_files=total_files,622 total_additions=total_additions,623 risk_level=risk_level,624 review_strategy=strategy625 )626 627 state["selected_model"] = selected_model628 629 logger.info(f"Route decision: strategy={strategy}, model={selected_model}, "630 f"files={total_files}, additions={total_additions}, risk={risk_level}")631 632 state["messages"].append({633 "role": "system",634 "content": f"Review strategy: {strategy} using {selected_model}"635 })636 637 except Exception as e:638 logger.error(f"Error in route node: {e}", exc_info=True)639 # Default to standard review on error640 state["review_strategy"] = "standard"641 state["selected_model"] = self.config.flash_model642 state["errors"] = state.get("errors", []) + [643 {"message": str(e), "step": "route", "timestamp": datetime.utcnow().isoformat()}644 ]645 646 return state647 648 def _route_decision(self, state: ReviewState) -> str:649 """650 Conditional edge: Decide which review path to take651 652 Returns: "quick", "standard", "deep", or "error"653 """654 # Check for critical errors in analysis/routing655 errors = state.get("errors", [])656 if errors and any(e.get("step") in ["analyze", "route"] for e in errors):657 logger.error(f"[ROUTE_DECISION] Errors detected, routing to error handler")658 return "error"659 660 strategy = state.get("review_strategy", "standard")661 logger.info(f"[ROUTE_DECISION] Selected strategy: {strategy}")662 return strategy663 664 async def _review_quick_node(self, state: ReviewState) -> ReviewState:665 """666 Quick review for small, low-risk PRs667 668 Uses gemini-flash-lite with focused prompt for speed669 Reviews all files together in a single prompt670 """671 logger.info(f"[REVIEW_QUICK] Processing PR #{state['pr_number']}")672 state["status"] = "reviewing"673 state["current_step"] = "quick_review"674 675 try:676 pr_context = state.get("pr_context", {})677 678 # Build quick review prompt with GraphRAG context679 issues_summary = self._format_issues_summary(pr_context)680 files_summary = self._format_files_summary(pr_context, max_files=10)681 entities_summary = self._format_entities_summary(state)682 dependencies_summary = self._format_dependencies_summary(state)683 similar_code_summary = self._format_similar_code(pr_context, state)684 685 prompt = self.gemini_client.templates.QUICK_REVIEW_PROMPT.format(686 pr_title=state["pr_title"],687 total_files=pr_context.get("total_files", 0),688 additions=pr_context.get("total_additions", 0),689 deletions=pr_context.get("total_deletions", 0),690 description=state.get("pr_description", "No description provided"),691 files_summary=files_summary,692 issues_summary=issues_summary,693 entities=entities_summary,694 dependencies=dependencies_summary,695 similar_code=similar_code_summary696 )697 698 model = state.get("selected_model", self.config.flash_lite_model)699 700 # M6: structured generation + citation validation701 review_output: ReviewOutput = await self.gemini_client.generate_structured_review(702 model_name=model,703 prompt=prompt,704 )705 valid_uids = self._collect_entity_uids(state)706 review_output = self.review_validator.validate_graph_citations(review_output, valid_uids)707 708 markdown = ReviewFormatter().format_structured_review(review_output)709 state["structured_review"] = review_output.model_dump()710 state["overall_summary"] = markdown711 state["processed_files"] = pr_context.get("total_files", 0)712 713 unverified = sum(1 for f in review_output.findings if not f.verified)714 logger.info(715 f"Quick review completed ({len(markdown)} chars, "716 f"{len(review_output.findings)} findings, {unverified} unverified)"717 )718 state["messages"].append({719 "role": "assistant",720 "content": (721 f"Quick review generated for {pr_context.get('total_files', 0)} files "722 f"({len(review_output.findings)} findings, {unverified} unverified citations)"723 ),724 })725 726 except Exception as e:727 logger.error(f"Error in quick review: {e}", exc_info=True)728 state["errors"] = state.get("errors", []) + [729 {"message": str(e), "step": "review_quick", "timestamp": datetime.utcnow().isoformat()}730 ]731 state["overall_summary"] = "Unable to generate quick review due to an error."732 733 return state734 735 async def _review_standard_node(self, state: ReviewState) -> ReviewState:736 """737 Standard review for typical PRs738 739 Reviews files in batches, prioritizing high-risk files740 Generates individual file reviews then aggregates741 """742 logger.info(f"[REVIEW_STANDARD] Processing PR #{state['pr_number']}")743 state["status"] = "reviewing"744 state["current_step"] = "standard_review"745 746 try:747 pr_context = state.get("pr_context", {})748 files = pr_context.get("files", [])749 750 # Prioritize files for review751 high_priority = state.get("high_priority_files", [])752 medium_priority = state.get("medium_priority_files", [])753 low_priority = state.get("low_priority_files", [])754 755 prioritized_files = high_priority + medium_priority + low_priority756 757 # Review files in parallel for much better performance758 file_reviews = {}759 model = state.get("selected_model", self.config.flash_model)760 761 # Collect files to review762 review_tasks = []763 files_to_review = []764 for filename in prioritized_files[:self.config.max_files_per_batch * 2]:765 file_ctx = self._find_file_context(files, filename)766 if file_ctx:767 files_to_review.append(filename)768 review_tasks.append(self._review_single_file(769 model=model,770 file_ctx=file_ctx,771 state=state772 ))773 774 # Review all files in parallel775 logger.info(f"Reviewing {len(review_tasks)} files in parallel...")776 import asyncio777 reviews = await asyncio.gather(*review_tasks, return_exceptions=True)778 779 # Collect results780 for filename, review in zip(files_to_review, reviews):781 if isinstance(review, Exception):782 logger.error(f"Error reviewing {filename}: {review}")783 file_reviews[filename] = {784 "filename": filename,785 "summary": f"Error: {str(review)}",786 "error": True787 }788 else:789 file_reviews[filename] = review790 791 state["file_reviews"] = file_reviews792 state["processed_files"] = len(file_reviews)793 794 logger.info(f"Standard review completed: {len(file_reviews)} files reviewed")795 796 state["messages"].append({797 "role": "assistant",798 "content": f"Reviewed {len(file_reviews)} files in standard mode"799 })800 801 except Exception as e:802 logger.error(f"Error in standard review: {e}", exc_info=True)803 state["errors"] = state.get("errors", []) + [804 {"message": str(e), "step": "review_standard", "timestamp": datetime.utcnow().isoformat()}805 ]806 807 return state808 809 async def _review_deep_node(self, state: ReviewState) -> ReviewState:810 """811 Deep review for complex, high-risk PRs812 813 Uses gemini-pro for thorough analysis814 Reviews all files individually with full context815 Includes architecture and security deep dives816 """817 logger.info(f"[REVIEW_DEEP] Processing PR #{state['pr_number']}")818 state["status"] = "reviewing"819 state["current_step"] = "deep_review"820 821 try:822 pr_context = state.get("pr_context", {})823 files = pr_context.get("files", [])824 825 # Build comprehensive deep review prompt826 critical_issues = self._format_critical_issues(pr_context)827 high_issues = self._format_high_issues(pr_context)828 files_details = self._format_files_details(files)829 830 prompt = self.gemini_client.templates.DEEP_REVIEW_PROMPT.format(831 pr_title=state["pr_title"],832 description=state.get("pr_description", "No description provided"),833 total_files=pr_context.get("total_files", 0),834 additions=pr_context.get("total_additions", 0),835 deletions=pr_context.get("total_deletions", 0),836 languages=", ".join(pr_context.get("languages", [])),837 risk_level=pr_context.get("risk_level", "unknown").upper(),838 critical_issues=critical_issues,839 high_issues=high_issues,840 affected_callers=pr_context.get("affected_callers", 0),841 complexity_hotspots=self._format_complexity_hotspots(pr_context),842 coupling_files=self._format_coupling_files(pr_context),843 similar_code=self._format_similar_code(pr_context, state),844 entities=self._format_entities_summary(state),845 dependencies=self._format_dependencies_summary(state),846 files_details=files_details847 )848 849 model = state.get("selected_model", self.config.pro_model)850 851 # M6: structured generation + citation validation852 logger.info(f"Generating structured deep review with {model}...")853 review_output: ReviewOutput = await self.gemini_client.generate_structured_review(854 model_name=model,855 prompt=prompt,856 )857 valid_uids = self._collect_entity_uids(state)858 review_output = self.review_validator.validate_graph_citations(review_output, valid_uids)859 860 markdown = ReviewFormatter().format_structured_review(review_output)861 state["structured_review"] = review_output.model_dump()862 state["overall_summary"] = markdown863 state["processed_files"] = len(files)864 865 unverified = sum(1 for f in review_output.findings if not f.verified)866 logger.info(867 f"Deep review completed ({len(markdown)} chars, "868 f"{len(review_output.findings)} findings, {unverified} unverified)"869 )870 state["messages"].append({871 "role": "assistant",872 "content": (873 f"Deep review generated with comprehensive analysis "874 f"({len(review_output.findings)} findings, {unverified} unverified citations)"875 ),876 })877 878 except Exception as e:879 logger.error(f"Error in deep review: {e}", exc_info=True)880 state["errors"] = state.get("errors", []) + [881 {"message": str(e), "step": "review_deep", "timestamp": datetime.utcnow().isoformat()}882 ]883 state["overall_summary"] = "Unable to generate deep review due to an error."884 885 return state886 887 async def _aggregate_node(self, state: ReviewState) -> ReviewState:888 """889 Aggregate results and create final review890 891 For quick/deep reviews: formats the single review892 For standard reviews: aggregates multiple file reviews into unified review893 """894 logger.info(f"[AGGREGATE] Aggregating review for PR #{state['pr_number']}")895 896 state["status"] = "aggregating"897 state["current_step"] = "aggregation"898 899 try:900 review_strategy = state.get("review_strategy", "standard")901 902 # If we already have overall_summary (quick/deep), just format it903 if state.get("overall_summary"):904 logger.info("Using existing overall summary")905 state["status"] = "completed"906 state["completed_at"] = datetime.utcnow().isoformat()907 return state908 909 # For standard review, aggregate file reviews910 file_reviews = state.get("file_reviews", {})911 912 if not file_reviews:913 logger.warning("No file reviews to aggregate")914 state["overall_summary"] = "No detailed reviews were generated."915 state["status"] = "completed"916 state["completed_at"] = datetime.utcnow().isoformat()917 return state918 919 # Build aggregation prompt920 pr_context = state.get("pr_context", {})921 file_reviews_text = "\n\n".join([922 f"### {filename}\n{review.get('summary', '')}" 923 for filename, review in file_reviews.items()924 ])925 926 prompt = self.gemini_client.templates.AGGREGATION_PROMPT.format(927 pr_title=state["pr_title"],928 files_count=len(file_reviews),929 total_issues=len(pr_context.get("critical_issues", [])) + len(pr_context.get("high_issues", [])),930 file_reviews=file_reviews_text,931 critical_count=len(pr_context.get("critical_issues", [])),932 high_count=len(pr_context.get("high_issues", [])),933 medium_count=len(pr_context.get("medium_issues", []))934 )935 936 model = state.get("selected_model", self.config.flash_model)937 938 # M6: structured aggregation + citation validation939 logger.info("Generating structured aggregated review...")940 review_output: ReviewOutput = await self.gemini_client.generate_structured_review(941 model_name=model,942 prompt=prompt,943 )944 valid_uids = self._collect_entity_uids(state)945 review_output = self.review_validator.validate_graph_citations(review_output, valid_uids)946 947 markdown = ReviewFormatter().format_structured_review(review_output)948 state["structured_review"] = review_output.model_dump()949 state["overall_summary"] = markdown950 state["review_summary"] = {951 "total_issues": sum(len(f.get("issues", [])) for f in file_reviews.values()),952 "total_suggestions": sum(1 for f in file_reviews.values() for i in f.get("issues", []) if i.get("category") == "suggestion"),953 "critical_count": sum(1 for f in file_reviews.values() for i in f.get("issues", []) if i.get("severity") == "critical"),954 "high_count": sum(1 for f in file_reviews.values() for i in f.get("issues", []) if i.get("severity") == "high"),955 "medium_count": sum(1 for f in file_reviews.values() for i in f.get("issues", []) if i.get("severity") == "medium"),956 "low_count": sum(1 for f in file_reviews.values() for i in f.get("issues", []) if i.get("severity") == "low"),957 "parse_failure_total": state.get("parse_failure_total", 0)958 }959 960 unverified = sum(1 for f in review_output.findings if not f.verified)961 logger.info(962 f"Aggregation completed ({len(markdown)} chars, "963 f"{len(review_output.findings)} findings, {unverified} unverified)"964 )965 state["messages"].append({966 "role": "assistant",967 "content": (968 f"Final review aggregated from individual file reviews "969 f"({len(review_output.findings)} findings, {unverified} unverified citations)"970 ),971 })972 973 except Exception as e:974 logger.error(f"Error in aggregation: {e}", exc_info=True)975 state["errors"] = state.get("errors", []) + [976 {"message": str(e), "step": "aggregate", "timestamp": datetime.utcnow().isoformat()}977 ]978 # Fallback: concatenate file reviews979 file_reviews = state.get("file_reviews", {})980 if file_reviews:981 state["overall_summary"] = "\n\n".join([982 f"**{filename}**\n{review.get('summary', '')}" 983 for filename, review in file_reviews.items()984 ])985 else:986 state["overall_summary"] = "Unable to generate aggregated review."987 988 state["status"] = "completed"989 state["completed_at"] = datetime.utcnow().isoformat()990 991 return state992 993 async def _error_handler_node(self, state: ReviewState) -> ReviewState:994 """995 Handle errors and determine retry strategy996 997 Recovery strategies:998 1. Transient errors (rate limit, network): Retry999 2. Authentication errors: Fail immediately1000 3. Review generation errors: Try fallback model1001 4. Max retries exceeded: Generate basic review from context1002 """1003 logger.error(f"[ERROR_HANDLER] Processing error for PR #{state['pr_number']}")1004 1005 state["status"] = "error_recovery"1006 retry_count = state.get("retry_count", 0)1007 state["retry_count"] = retry_count + 11008 1009 errors = state.get("errors", [])1010 if not errors:1011 logger.warning("Error handler called but no errors found")1012 return state1013 1014 last_error = errors[-1]1015 error_msg = last_error.get("message", "")1016 error_step = last_error.get("step", "unknown")1017 1018 logger.info(f"Handling error from step '{error_step}': {error_msg}")1019 1020 # Check if error is recoverable1021 if "rate limit" in error_msg.lower():1022 logger.info("Rate limit error detected - will retry with delay")1023 import asyncio1024 await asyncio.sleep(self.config.retry_delay_seconds * (retry_count + 1))1025 return state1026 1027 elif "authentication" in error_msg.lower() or "api key" in error_msg.lower():1028 logger.error("Authentication error - cannot retry")1029 state["status"] = "failed"1030 state["overall_summary"] = "Unable to generate review: Authentication error with Gemini API"1031 return state1032 1033 elif retry_count >= self.config.max_retries:1034 logger.error(f"Max retries ({self.config.max_retries}) exceeded")1035 # Generate fallback review from context1036 state = await self._generate_fallback_review(state)1037 return state1038 1039 else:1040 logger.info(f"Recoverable error - will retry (attempt {retry_count + 1}/{self.config.max_retries})")1041 return state1042 1043 def _error_decision(self, state: ReviewState) -> str:1044 """1045 Conditional edge: Decide whether to retry or fail1046 1047 Returns "retry" or "end"1048 """1049 retry_count = state.get("retry_count", 0)1050 status = state.get("status", "")1051 1052 # If we already have a fallback review or explicitly failed, end1053 if status == "failed" or state.get("overall_summary"):1054 logger.info(f"[ERROR_DECISION] Ending workflow (status={status})")1055 return "end"1056 1057 if retry_count < self.config.max_retries:1058 logger.info(f"[ERROR_DECISION] Retrying (attempt {retry_count + 1}/{self.config.max_retries})")1059 return "retry"1060 else:1061 logger.error(f"[ERROR_DECISION] Max retries exceeded, ending workflow")1062 return "end"1063 1064 # ========================================================================1065 # HELPER METHODS (Phase 4.4)1066 # ========================================================================1067 1068 async def _quick_scan_file(self, file_ctx: Dict, state: ReviewState) -> Dict[str, Any]:1069 """Phase 3: Quick scan for critical issues (Pass 1)1070 1071 Fast scan to identify:1072 - Security vulnerabilities1073 - Critical bugs1074 - Immediate blockers1075 1076 Returns dict with critical_issues flag and quick summary1077 """1078 try:1079 filename = file_ctx.get("filename", "unknown")1080 logger.info(f"[PHASE3] Quick scanning {filename}...")1081 1082 # Get file diff1083 diff = self._get_file_diff(state["files"], filename)1084 formatted_diff = self.format_diff_for_review(diff, filename)1085 1086 # Quick scan prompt - focused on critical issues only1087 prompt = self.gemini_client.templates.QUICK_SCAN_PROMPT.format(1088 filename=filename,1089 language=file_ctx.get("language", "unknown"),1090 additions=file_ctx.get("additions", 0),1091 deletions=file_ctx.get("deletions", 0),1092 diff=formatted_diff1093 )1094 1095 # Use faster model for quick scan1096 scan_result = await self.gemini_client.generate_review(1097 model_name="gemini-2.0-flash-exp", # Fast model1098 prompt=prompt1099 )1100 1101 # Parse for critical issues1102 has_critical = any([1103 "🔴" in scan_result,1104 "CRITICAL" in scan_result.upper(),1105 "SECURITY" in scan_result.upper(),1106 "VULNERABILITY" in scan_result.upper()1107 ])1108 1109 logger.info(f"[PHASE3] Quick scan complete for {filename} (critical: {has_critical})")1110 1111 return {1112 "filename": filename,1113 "has_critical": has_critical,1114 "quick_summary": scan_result,1115 "scan_time": "quick"1116 }1117 1118 except Exception as e:1119 logger.error(f"[PHASE3] Error in quick scan for {file_ctx.get('filename')}: {e}")1120 return {1121 "filename": file_ctx.get("filename", "unknown"),1122 "has_critical": False, # Assume no critical to avoid blocking1123 "quick_summary": f"Quick scan error: {str(e)}",1124 "error": True1125 }1126 1127 async def _detailed_review_file(self, model: str, file_ctx: Dict, state: ReviewState, quick_scan: Dict) -> Dict[str, Any]:1128 """Phase 3: Detailed review with GraphRAG (Pass 2)1129 1130 Deep analysis including:1131 - All issue levels1132 - GraphRAG context integration1133 - Similar code patterns1134 - Dependency impact1135 - Refactoring opportunities1136 """1137 try:1138 filename = file_ctx.get("filename", "unknown")1139 logger.info(f"[PHASE3] Detailed review for {filename} (has_critical: {quick_scan.get('has_critical')})")1140 1141 # Format issues summary1142 issues_summary = file_ctx.get("issues_summary", {})1143 if isinstance(issues_summary, dict) and all(isinstance(k, str) for k in issues_summary.keys()):1144 issue_parts = []1145 for severity in ["critical", "high", "medium", "low"]:1146 count = issues_summary.get(severity, 0)1147 if count > 0:1148 issue_parts.append(f"{severity.upper()}: {count}")1149 issues = ", ".join(issue_parts) if issue_parts else "No issues detected"1150 else:1151 issues = "No issues detected"1152 1153 # Handle dependencies1154 deps = file_ctx.get("dependencies", [])1155 if isinstance(deps, list) and deps:1156 dependencies = "\n".join([f"- {dep}" for dep in deps[:10]])1157 else:1158 dependencies = "None"1159 1160 # Extract similar code from context (GraphRAG!) - ENHANCED1161 similar_code = self._format_similar_code_for_file_enhanced(file_ctx, state)1162 1163 # Format entities with relationships - NEW1164 entities = self._format_entities_for_file(file_ctx, state)1165 1166 # Get file diff1167 diff = self._get_file_diff(state["files"], filename)1168 formatted_diff = self.format_diff_for_review(diff, filename)1169 1170 # Phase 3: Include quick scan findings in detailed review1171 prompt = self.gemini_client.templates.FILE_REVIEW_PROMPT.format(1172 filename=filename,1173 language=file_ctx.get("language", "unknown"),1174 additions=file_ctx.get("additions", 0),1175 deletions=file_ctx.get("deletions", 0),1176 issues=issues,1177 similar_code=similar_code,1178 dependencies=dependencies,1179 entities=entities, # NEW1180 diff=formatted_diff1181 )1182 1183 # M6: structured file review + citation validation1184 review_output: ReviewOutput = await self.gemini_client.generate_structured_review(1185 model_name=model,1186 prompt=prompt,1187 )1188 valid_uids = self._collect_entity_uids(state)1189 review_output = self.review_validator.validate_graph_citations(review_output, valid_uids)1190 review_text = ReviewFormatter().format_structured_review(review_output)1191 1192 # Phase 4: Validate rendered review text against actual diff1193 files_for_validation = [{1194 "filename": filename,1195 "patch": diff1196 }]1197 is_valid, validation_warnings, validation_metrics = self.review_validator.validate_review(1198 review_text,1199 files_for_validation1200 )