diff --git a/apps/backend/cli/main.py b/apps/backend/cli/main.py index cfb6a6a4..0b4db55c 100644 --- a/apps/backend/cli/main.py +++ b/apps/backend/cli/main.py @@ -288,6 +288,28 @@ def main() -> None: # Set up environment first setup_environment() + # Initialize Sentry early to capture any startup errors + from core.sentry import capture_exception, init_sentry + + init_sentry(component="cli") + + try: + _run_cli() + except KeyboardInterrupt: + # Clean exit on Ctrl+C + sys.exit(130) + except Exception as e: + # Capture unexpected errors to Sentry + capture_exception(e) + print(f"\nUnexpected error: {e}") + sys.exit(1) + + +def _run_cli() -> None: + """Run the CLI logic (extracted for error handling).""" + # Import here to avoid import errors during startup + from core.sentry import set_context + # Parse arguments args = parse_args() @@ -358,6 +380,15 @@ def main() -> None: debug_success("run.py", "Spec found", spec_dir=str(spec_dir)) + # Set Sentry context for error tracking + set_context( + "spec", + { + "name": spec_dir.name, + "project": str(project_dir), + }, + ) + # Handle build management commands if args.merge_preview: from cli.workspace_commands import handle_merge_preview_command diff --git a/apps/backend/core/io_utils.py b/apps/backend/core/io_utils.py new file mode 100644 index 00000000..c5a8a155 --- /dev/null +++ b/apps/backend/core/io_utils.py @@ -0,0 +1,94 @@ +""" +I/O Utilities for Safe Console Output +===================================== + +Safe I/O operations for processes running as subprocesses. + +When the backend runs as a subprocess of the Electron app, the parent +process may close the pipe at any time (e.g., user closes the app, +process killed, etc.). This module provides utilities to handle these +cases gracefully. +""" + +from __future__ import annotations + +import logging +import sys + +logger = logging.getLogger(__name__) + +# Track if pipe is broken to avoid repeated failed writes +_pipe_broken = False + + +def safe_print(message: str, flush: bool = True) -> None: + """ + Print to stdout with BrokenPipeError handling. + + When running as a subprocess (e.g., from Electron), the parent process + may close the pipe at any time. This function gracefully handles that + case instead of raising an exception. + + Args: + message: The message to print + flush: Whether to flush stdout after printing (default True) + """ + global _pipe_broken + + # Skip if we already know the pipe is broken + if _pipe_broken: + return + + try: + print(message, flush=flush) + except BrokenPipeError: + # Pipe closed by parent process - this is expected during shutdown + _pipe_broken = True + # Quietly close stdout to prevent further errors + try: + sys.stdout.close() + except Exception: + pass + logger.debug("Output pipe closed by parent process") + except ValueError as e: + # Handle writes to closed file (can happen after stdout.close()) + if "closed file" in str(e).lower(): + _pipe_broken = True + logger.debug("Output stream closed") + else: + # Re-raise unexpected ValueErrors + raise + except OSError as e: + # Handle other pipe-related errors (EPIPE, etc.) + if e.errno == 32: # EPIPE - Broken pipe + _pipe_broken = True + try: + sys.stdout.close() + except Exception: + pass + logger.debug("Output pipe closed (EPIPE)") + else: + # Re-raise unexpected OS errors + raise + + +def is_pipe_broken() -> bool: + """Check if the output pipe has been closed.""" + return _pipe_broken + + +def reset_pipe_state() -> None: + """ + Reset pipe broken state. + + Useful for testing or when starting a new subprocess context where + stdout has been reopened. Should only be called when stdout is known + to be functional (e.g., in a fresh subprocess with a new stdout). + + Warning: + Calling this after stdout has been closed will result in safe_print() + attempting to write to the closed stream. The ValueError will be + caught and the pipe will be marked as broken again. + """ + global _pipe_broken + _pipe_broken = False diff --git a/apps/backend/core/sentry.py b/apps/backend/core/sentry.py new file mode 100644 index 00000000..0634f4ad --- /dev/null +++ b/apps/backend/core/sentry.py @@ -0,0 +1,406 @@ +""" +Sentry Error Tracking for Python Backend +========================================= + +Initializes Sentry for the Python backend with: +- Privacy-preserving path masking (usernames removed) +- Release tracking matching the Electron frontend +- Environment variable configuration (same as frontend) + +Configuration: +- SENTRY_DSN: Required to enable Sentry (same as frontend) +- SENTRY_TRACES_SAMPLE_RATE: Performance monitoring sample rate (0-1, default: 0.1) +- SENTRY_ENVIRONMENT: Override environment (default: auto-detected) + +Privacy Note: +- Usernames are masked from all file paths +- Project paths remain visible for debugging (this is expected) +- No user identifiers are collected +""" + +from __future__ import annotations + +import logging +import os +import re +import sys +from pathlib import Path +from typing import Any + +logger = logging.getLogger(__name__) + +# Track initialization state +_sentry_initialized = False +_sentry_enabled = False + +# Production trace sample rate (10%) +PRODUCTION_TRACE_SAMPLE_RATE = 0.1 + + +def _get_version() -> str: + """ + Get the application version. + + Tries to read from package.json in the frontend directory, + falling back to a default version. + """ + try: + # Try to find package.json relative to this file + backend_dir = Path(__file__).parent.parent + frontend_dir = backend_dir.parent / "frontend" + package_json = frontend_dir / "package.json" + + if package_json.exists(): + import json + + with open(package_json) as f: + data = json.load(f) + return data.get("version", "0.0.0") + except Exception as e: + logger.debug(f"Version detection failed: {e}") + + return "0.0.0" + + +def _mask_user_paths(text: str) -> str: + """ + Mask user-specific paths for privacy. + + Replaces usernames in common OS path patterns: + - macOS: /Users/username/... becomes /Users/***/... + - Windows: C:\\Users\\username\\... becomes C:\\Users\\***\\... + - Linux: /home/username/... becomes /home/***/... + - WSL: /mnt/c/Users/username/... becomes /mnt/c/Users/***/... + + Note: Project paths remain visible for debugging purposes. + """ + if not text: + return text + + # macOS: /Users/username/... + text = re.sub(r"/Users/[^/]+(?=/|$)", "/Users/***", text) + + # Windows: C:\Users\username\... + text = re.sub( + r"[A-Za-z]:\\Users\\[^\\]+(?=\\|$)", + lambda m: f"{m.group(0)[0]}:\\Users\\***", + text, + ) + + # Linux: /home/username/... + text = re.sub(r"/home/[^/]+(?=/|$)", "/home/***", text) + + # WSL: /mnt/c/Users/username/... (accessing Windows filesystem from WSL) + text = re.sub( + r"/mnt/[a-z]/Users/[^/]+(?=/|$)", + lambda m: f"{m.group(0)[:6]}/Users/***", + text, + ) + + return text + + +def _mask_object_paths(obj: Any, _depth: int = 0) -> Any: + """ + Recursively mask paths in an object. + + Args: + obj: The object to mask paths in + _depth: Current recursion depth (internal use) + + Returns: + Object with paths masked + """ + # Prevent stack overflow on deeply nested or circular structures + if _depth > 50: + return obj + + if obj is None: + return obj + + if isinstance(obj, str): + return _mask_user_paths(obj) + + if isinstance(obj, list): + return [_mask_object_paths(item, _depth + 1) for item in obj] + + if isinstance(obj, dict): + return { + key: _mask_object_paths(value, _depth + 1) for key, value in obj.items() + } + + return obj + + +def _before_send(event: dict, hint: dict) -> dict | None: + """ + Process event before sending to Sentry. + + Applies privacy masking to all paths in the event. + """ + if not _sentry_enabled: + return None + + # Mask paths in exception stack traces + if "exception" in event and "values" in event["exception"]: + for exception in event["exception"]["values"]: + if "stacktrace" in exception and "frames" in exception["stacktrace"]: + for frame in exception["stacktrace"]["frames"]: + if "filename" in frame: + frame["filename"] = _mask_user_paths(frame["filename"]) + if "abs_path" in frame: + frame["abs_path"] = _mask_user_paths(frame["abs_path"]) + if "value" in exception: + exception["value"] = _mask_user_paths(exception["value"]) + + # Mask paths in breadcrumbs + if "breadcrumbs" in event: + for breadcrumb in event.get("breadcrumbs", {}).get("values", []): + if "message" in breadcrumb: + breadcrumb["message"] = _mask_user_paths(breadcrumb["message"]) + if "data" in breadcrumb: + breadcrumb["data"] = _mask_object_paths(breadcrumb["data"]) + + # Mask paths in message + if "message" in event: + event["message"] = _mask_user_paths(event["message"]) + + # Mask paths in tags + if "tags" in event: + event["tags"] = _mask_object_paths(event["tags"]) + + # Mask paths in contexts + if "contexts" in event: + event["contexts"] = _mask_object_paths(event["contexts"]) + + # Mask paths in extra data + if "extra" in event: + event["extra"] = _mask_object_paths(event["extra"]) + + # Clear user info for privacy + if "user" in event: + event["user"] = {} + + return event + + +def init_sentry( + component: str = "backend", + force_enable: bool = False, +) -> bool: + """ + Initialize Sentry for the Python backend. + + Args: + component: Component name for tagging (e.g., "backend", "github-runner") + force_enable: Force enable even without packaged app detection + + Returns: + True if Sentry was initialized, False otherwise + """ + global _sentry_initialized, _sentry_enabled + + if _sentry_initialized: + return _sentry_enabled + + _sentry_initialized = True + + # Get DSN from environment variable + dsn = os.environ.get("SENTRY_DSN", "") + + if not dsn: + logger.debug("[Sentry] No SENTRY_DSN configured - error reporting disabled") + return False + + # Check if we should enable Sentry + # Enable if: + # - Running from packaged app (detected by __compiled__ or frozen) + # - SENTRY_DEV=true is set + # - force_enable is True + is_packaged = getattr(sys, "frozen", False) or hasattr(sys, "__compiled__") + sentry_dev = os.environ.get("SENTRY_DEV", "").lower() in ("true", "1", "yes") + should_enable = is_packaged or sentry_dev or force_enable + + if not should_enable: + logger.debug( + "[Sentry] Development mode - error reporting disabled (set SENTRY_DEV=true to enable)" + ) + return False + + try: + import sentry_sdk + from sentry_sdk.integrations.logging import LoggingIntegration + except ImportError: + logger.warning("[Sentry] sentry-sdk not installed - error reporting disabled") + return False + + # Get configuration from environment variables + version = _get_version() + environment = os.environ.get( + "SENTRY_ENVIRONMENT", "production" if is_packaged else "development" + ) + + # Get sample rates + traces_sample_rate = PRODUCTION_TRACE_SAMPLE_RATE + try: + env_rate = os.environ.get("SENTRY_TRACES_SAMPLE_RATE") + if env_rate: + parsed = float(env_rate) + if 0 <= parsed <= 1: + traces_sample_rate = parsed + except (ValueError, TypeError): + pass + + # Configure logging integration to capture errors and warnings + logging_integration = LoggingIntegration( + level=logging.INFO, # Capture INFO and above as breadcrumbs + event_level=logging.ERROR, # Send ERROR and above as events + ) + + # Initialize Sentry + sentry_sdk.init( + dsn=dsn, + environment=environment, + release=f"auto-claude@{version}", + traces_sample_rate=traces_sample_rate, + before_send=_before_send, + integrations=[logging_integration], + # Don't send PII + send_default_pii=False, + ) + + # Set component tag + sentry_sdk.set_tag("component", component) + + _sentry_enabled = True + logger.info( + f"[Sentry] Backend initialized (component: {component}, release: auto-claude@{version}, traces: {traces_sample_rate})" + ) + + return True + + +def capture_exception(error: Exception, **kwargs) -> None: + """ + Capture an exception and send to Sentry. + + Safe to call even if Sentry is not initialized. + + Args: + error: The exception to capture + **kwargs: Additional context to attach to the event + """ + if not _sentry_enabled: + logger.error(f"[Sentry] Not enabled, exception not captured: {error}") + return + + try: + import sentry_sdk + + with sentry_sdk.push_scope() as scope: + for key, value in kwargs.items(): + # Apply defensive path masking for extra data + masked_value = ( + _mask_object_paths(value) + if isinstance(value, (str, dict, list)) + else value + ) + scope.set_extra(key, masked_value) + sentry_sdk.capture_exception(error) + except ImportError: + logger.error(f"[Sentry] SDK not installed, exception not captured: {error}") + except Exception as e: + logger.error(f"[Sentry] Failed to capture exception: {e}") + + +def capture_message(message: str, level: str = "info", **kwargs) -> None: + """ + Capture a message and send to Sentry. + + Safe to call even if Sentry is not initialized. + + Args: + message: The message to capture + level: Log level (debug, info, warning, error, fatal) + **kwargs: Additional context to attach to the event + """ + if not _sentry_enabled: + return + + try: + import sentry_sdk + + with sentry_sdk.push_scope() as scope: + for key, value in kwargs.items(): + # Apply defensive path masking for extra data (same as capture_exception) + masked_value = ( + _mask_object_paths(value) + if isinstance(value, (str, dict, list)) + else value + ) + scope.set_extra(key, masked_value) + sentry_sdk.capture_message(message, level=level) + except ImportError: + logger.debug("[Sentry] SDK not installed") + except Exception as e: + logger.error(f"[Sentry] Failed to capture message: {e}") + + +def set_context(name: str, data: dict) -> None: + """ + Set context data for subsequent events. + + Safe to call even if Sentry is not initialized. + + Args: + name: Context name (e.g., "pr_review", "spec") + data: Context data dictionary + """ + if not _sentry_enabled: + return + + try: + import sentry_sdk + + # Apply path masking to context data before sending to Sentry + masked_data = _mask_object_paths(data) + sentry_sdk.set_context(name, masked_data) + except ImportError: + logger.debug("[Sentry] SDK not installed") + except Exception as e: + logger.debug(f"Failed to set context '{name}': {e}") + + +def set_tag(key: str, value: str) -> None: + """ + Set a tag for subsequent events. + + Safe to call even if Sentry is not initialized. + + Args: + key: Tag key + value: Tag value + """ + if not _sentry_enabled: + return + + try: + import sentry_sdk + + # Apply path masking to tag value + masked_value = _mask_user_paths(value) if isinstance(value, str) else value + sentry_sdk.set_tag(key, masked_value) + except ImportError: + logger.debug("[Sentry] SDK not installed") + except Exception as e: + logger.debug(f"Failed to set tag '{key}': {e}") + + +def is_enabled() -> bool: + """Check if Sentry is enabled.""" + return _sentry_enabled + + +def is_initialized() -> bool: + """Check if Sentry initialization has been attempted.""" + return _sentry_initialized diff --git a/apps/backend/requirements.txt b/apps/backend/requirements.txt index 95c8a1ea..13a9e1fb 100644 --- a/apps/backend/requirements.txt +++ b/apps/backend/requirements.txt @@ -19,3 +19,6 @@ google-generativeai>=0.8.0 # Pydantic for structured output schemas pydantic>=2.0.0 + +# Error tracking (optional - requires SENTRY_DSN environment variable) +sentry-sdk>=2.0.0 diff --git a/apps/backend/runners/github/context_gatherer.py b/apps/backend/runners/github/context_gatherer.py index 9a3c5512..535298ee 100644 --- a/apps/backend/runners/github/context_gatherer.py +++ b/apps/backend/runners/github/context_gatherer.py @@ -25,8 +25,10 @@ from typing import TYPE_CHECKING try: from .gh_client import GHClient, PRTooLargeError + from .services.io_utils import safe_print except (ImportError, ValueError, SystemError): from gh_client import GHClient, PRTooLargeError + from services.io_utils import safe_print # Validation patterns for git refs and paths (defense-in-depth) # These patterns allow common valid characters while rejecting potentially dangerous ones @@ -232,11 +234,11 @@ class PRContextGatherer: Returns: PRContext with all necessary information for review """ - print(f"[Context] Gathering context for PR #{self.pr_number}...", flush=True) + safe_print(f"[Context] Gathering context for PR #{self.pr_number}...") # Fetch basic PR metadata pr_data = await self._fetch_pr_metadata() - print( + safe_print( f"[Context] PR metadata: {pr_data['title']} by {pr_data['author']['login']}", flush=True, ) @@ -248,7 +250,7 @@ class PRContextGatherer: if head_sha and base_sha: refs_available = await self._ensure_pr_refs_available(head_sha, base_sha) if not refs_available: - print( + safe_print( "[Context] Warning: Could not fetch PR refs locally. " "Will use GitHub API patches as fallback.", flush=True, @@ -256,27 +258,27 @@ class PRContextGatherer: # Fetch changed files with content changed_files = await self._fetch_changed_files(pr_data) - print(f"[Context] Fetched {len(changed_files)} changed files", flush=True) + safe_print(f"[Context] Fetched {len(changed_files)} changed files") # Fetch full diff diff = await self._fetch_pr_diff() - print(f"[Context] Fetched diff: {len(diff)} chars", flush=True) + safe_print(f"[Context] Fetched diff: {len(diff)} chars") # Detect repo structure repo_structure = self._detect_repo_structure() - print("[Context] Detected repo structure", flush=True) + safe_print("[Context] Detected repo structure") # Find related files related_files = self._find_related_files(changed_files) - print(f"[Context] Found {len(related_files)} related files", flush=True) + safe_print(f"[Context] Found {len(related_files)} related files") # Fetch commits commits = await self._fetch_commits() - print(f"[Context] Fetched {len(commits)} commits", flush=True) + safe_print(f"[Context] Fetched {len(commits)} commits") # Fetch AI bot comments for triage ai_bot_comments = await self._fetch_ai_bot_comments() - print(f"[Context] Fetched {len(ai_bot_comments)} AI bot comments", flush=True) + safe_print(f"[Context] Fetched {len(ai_bot_comments)} AI bot comments") # Check if diff was truncated (empty diff but files were changed) diff_truncated = len(diff) == 0 and len(changed_files) > 0 @@ -287,7 +289,7 @@ class PRContextGatherer: has_merge_conflicts = mergeable == "CONFLICTING" if has_merge_conflicts: - print( + safe_print( f"[Context] ⚠️ PR has merge conflicts (mergeStateStatus: {merge_state_status})", flush=True, ) @@ -356,12 +358,12 @@ class PRContextGatherer: """ # Validate SHAs before using in git commands if not _validate_git_ref(head_sha): - print( + safe_print( f"[Context] Invalid head SHA rejected: {head_sha[:50]}...", flush=True ) return False if not _validate_git_ref(base_sha): - print( + safe_print( f"[Context] Invalid base SHA rejected: {base_sha[:50]}...", flush=True ) return False @@ -381,14 +383,14 @@ class PRContextGatherer: stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=30.0) if proc.returncode == 0: - print( + safe_print( f"[Context] Fetched PR refs: base={base_sha[:8]} → head={head_sha[:8]}", flush=True, ) return True else: # If direct SHA fetch fails, try fetching the PR ref - print("[Context] Direct SHA fetch failed, trying PR ref...", flush=True) + safe_print("[Context] Direct SHA fetch failed, trying PR ref...") proc2 = await asyncio.create_subprocess_exec( "git", "fetch", @@ -400,21 +402,21 @@ class PRContextGatherer: ) await asyncio.wait_for(proc2.communicate(), timeout=30.0) if proc2.returncode == 0: - print( + safe_print( f"[Context] Fetched PR ref: refs/pr/{self.pr_number}", flush=True, ) return True - print( + safe_print( f"[Context] Failed to fetch PR refs: {stderr.decode('utf-8')}", flush=True, ) return False except asyncio.TimeoutError: - print("[Context] Timeout fetching PR refs", flush=True) + safe_print("[Context] Timeout fetching PR refs") return False except Exception as e: - print(f"[Context] Error fetching PR refs: {e}", flush=True) + safe_print(f"[Context] Error fetching PR refs: {e}") return False async def _fetch_changed_files(self, pr_data: dict) -> list[ChangedFile]: @@ -435,7 +437,7 @@ class PRContextGatherer: additions = file_info.get("additions", 0) deletions = file_info.get("deletions", 0) - print(f"[Context] Processing {path} ({status})...", flush=True) + safe_print(f"[Context] Processing {path} ({status})...") # Use commit SHAs if available (works for fork PRs), fallback to branch names head_ref = pr_data.get("headRefOid") or pr_data["headRefName"] @@ -491,10 +493,10 @@ class PRContextGatherer: """ # Validate inputs to prevent command injection if not _validate_file_path(path): - print(f"[Context] Invalid file path rejected: {path[:50]}...", flush=True) + safe_print(f"[Context] Invalid file path rejected: {path[:50]}...") return "" if not _validate_git_ref(ref): - print(f"[Context] Invalid git ref rejected: {ref[:50]}...", flush=True) + safe_print(f"[Context] Invalid git ref rejected: {ref[:50]}...") return "" try: @@ -515,10 +517,10 @@ class PRContextGatherer: return stdout.decode("utf-8") except asyncio.TimeoutError: - print(f"[Context] Timeout reading {path} from {ref}", flush=True) + safe_print(f"[Context] Timeout reading {path} from {ref}") return "" except Exception as e: - print(f"[Context] Error reading {path} from {ref}: {e}", flush=True) + safe_print(f"[Context] Error reading {path} from {ref}: {e}") return "" async def _get_file_patch(self, path: str, base_ref: str, head_ref: str) -> str: @@ -535,15 +537,15 @@ class PRContextGatherer: """ # Validate inputs to prevent command injection if not _validate_file_path(path): - print(f"[Context] Invalid file path rejected: {path[:50]}...", flush=True) + safe_print(f"[Context] Invalid file path rejected: {path[:50]}...") return "" if not _validate_git_ref(base_ref): - print( + safe_print( f"[Context] Invalid base ref rejected: {base_ref[:50]}...", flush=True ) return "" if not _validate_git_ref(head_ref): - print( + safe_print( f"[Context] Invalid head ref rejected: {head_ref[:50]}...", flush=True ) return "" @@ -563,7 +565,7 @@ class PRContextGatherer: stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=10.0) if proc.returncode != 0: - print( + safe_print( f"[Context] Failed to get patch for {path}: {stderr.decode('utf-8')}", flush=True, ) @@ -571,10 +573,10 @@ class PRContextGatherer: return stdout.decode("utf-8") except asyncio.TimeoutError: - print(f"[Context] Timeout getting patch for {path}", flush=True) + safe_print(f"[Context] Timeout getting patch for {path}") return "" except Exception as e: - print(f"[Context] Error getting patch for {path}: {e}", flush=True) + safe_print(f"[Context] Error getting patch for {path}: {e}") return "" async def _fetch_pr_diff(self) -> str: @@ -587,8 +589,8 @@ class PRContextGatherer: try: return await self.gh_client.pr_diff(self.pr_number) except PRTooLargeError as e: - print(f"[Context] Warning: {str(e)}", flush=True) - print( + safe_print(f"[Context] Warning: {str(e)}") + safe_print( "[Context] Skipping full diff - will use individual file patches", flush=True, ) @@ -630,7 +632,7 @@ class PRContextGatherer: ai_comments.append(ai_comment) except Exception as e: - print(f"[Context] Error fetching AI bot comments: {e}", flush=True) + safe_print(f"[Context] Error fetching AI bot comments: {e}") return ai_comments @@ -698,7 +700,7 @@ class PRContextGatherer: return json.loads(result.stdout) return [] except Exception as e: - print(f"[Context] Error fetching review comments: {e}", flush=True) + safe_print(f"[Context] Error fetching review comments: {e}") return [] async def _fetch_pr_issue_comments(self) -> list[dict]: @@ -717,7 +719,7 @@ class PRContextGatherer: return json.loads(result.stdout) return [] except Exception as e: - print(f"[Context] Error fetching issue comments: {e}", flush=True) + safe_print(f"[Context] Error fetching issue comments: {e}") return [] def _detect_repo_structure(self) -> str: @@ -1015,7 +1017,7 @@ class FollowupContextGatherer: previous_sha = self.previous_review.reviewed_commit_sha if not previous_sha: - print( + safe_print( "[Followup] No reviewed_commit_sha in previous review, cannot gather incremental context", flush=True, ) @@ -1026,7 +1028,7 @@ class FollowupContextGatherer: current_commit_sha="", ) - print( + safe_print( f"[Followup] Gathering context since commit {previous_sha[:8]}...", flush=True, ) @@ -1035,7 +1037,7 @@ class FollowupContextGatherer: current_sha = await self.gh_client.get_pr_head_sha(self.pr_number) if not current_sha: - print("[Followup] Could not fetch current HEAD SHA", flush=True) + safe_print("[Followup] Could not fetch current HEAD SHA") return FollowupReviewContext( pr_number=self.pr_number, previous_review=self.previous_review, @@ -1044,7 +1046,7 @@ class FollowupContextGatherer: ) if previous_sha == current_sha: - print("[Followup] No new commits since last review", flush=True) + safe_print("[Followup] No new commits since last review") return FollowupReviewContext( pr_number=self.pr_number, previous_review=self.previous_review, @@ -1052,7 +1054,7 @@ class FollowupContextGatherer: current_commit_sha=current_sha, ) - print( + safe_print( f"[Followup] Comparing {previous_sha[:8]}...{current_sha[:8]}", flush=True ) @@ -1065,29 +1067,29 @@ class FollowupContextGatherer: pr_files, new_commits = await self.gh_client.get_pr_files_changed_since( self.pr_number, previous_sha, reviewed_file_blobs=reviewed_file_blobs ) - print( + safe_print( f"[Followup] PR has {len(pr_files)} files, " f"{len(new_commits)} commits since last review" + (" (blob comparison used)" if reviewed_file_blobs else ""), flush=True, ) except Exception as e: - print(f"[Followup] Error getting PR files/commits: {e}", flush=True) + safe_print(f"[Followup] Error getting PR files/commits: {e}") # Fallback to compare_commits if PR endpoints fail - print("[Followup] Falling back to commit comparison...", flush=True) + safe_print("[Followup] Falling back to commit comparison...") try: comparison = await self.gh_client.compare_commits( previous_sha, current_sha ) new_commits = comparison.get("commits", []) pr_files = comparison.get("files", []) - print( + safe_print( f"[Followup] Fallback: Found {len(new_commits)} commits, " f"{len(pr_files)} files (may include merge-introduced changes)", flush=True, ) except Exception as e2: - print(f"[Followup] Fallback also failed: {e2}", flush=True) + safe_print(f"[Followup] Fallback also failed: {e2}") return FollowupReviewContext( pr_number=self.pr_number, previous_review=self.previous_review, @@ -1099,7 +1101,7 @@ class FollowupContextGatherer: # Use PR files as the canonical list (excludes files from merged branches) commits = new_commits files = pr_files - print( + safe_print( f"[Followup] Found {len(commits)} new commits, {len(files)} changed files", flush=True, ) @@ -1123,7 +1125,7 @@ class FollowupContextGatherer: self.pr_number, self.previous_review.reviewed_at ) except Exception as e: - print(f"[Followup] Error fetching comments: {e}", flush=True) + safe_print(f"[Followup] Error fetching comments: {e}") comments = {"review_comments": [], "issue_comments": []} # Get formal PR reviews since last review (from Cursor, CodeRabbit, etc.) @@ -1132,7 +1134,7 @@ class FollowupContextGatherer: self.pr_number, self.previous_review.reviewed_at ) except Exception as e: - print(f"[Followup] Error fetching PR reviews: {e}", flush=True) + safe_print(f"[Followup] Error fetching PR reviews: {e}") pr_reviews = [] # Separate AI bot comments from contributor comments @@ -1179,7 +1181,7 @@ class FollowupContextGatherer: contributor_reviews ) - print( + safe_print( f"[Followup] Found {total_contributor_feedback} contributor feedback " f"({len(contributor_comments)} comments, {len(contributor_reviews)} reviews), " f"{total_ai_feedback} AI feedback " @@ -1200,12 +1202,12 @@ class FollowupContextGatherer: has_merge_conflicts = mergeable == "CONFLICTING" if has_merge_conflicts: - print( + safe_print( f"[Followup] ⚠️ PR has merge conflicts (mergeStateStatus: {merge_state_status})", flush=True, ) except Exception as e: - print(f"[Followup] Could not fetch merge status: {e}", flush=True) + safe_print(f"[Followup] Could not fetch merge status: {e}") return FollowupReviewContext( pr_number=self.pr_number, diff --git a/apps/backend/runners/github/orchestrator.py b/apps/backend/runners/github/orchestrator.py index 22d3e144..60ed9648 100644 --- a/apps/backend/runners/github/orchestrator.py +++ b/apps/backend/runners/github/orchestrator.py @@ -46,6 +46,7 @@ try: PRReviewEngine, TriageEngine, ) + from .services.io_utils import safe_print except (ImportError, ValueError, SystemError): # When imported directly (runner.py adds github dir to path) from bot_detection import BotDetector @@ -74,6 +75,7 @@ except (ImportError, ValueError, SystemError): PRReviewEngine, TriageEngine, ) + from services.io_utils import safe_print @dataclass @@ -267,12 +269,12 @@ class GitHubOrchestrator: comment_id=triage.comment_id, body=triage.response_comment, ) - print( + safe_print( f"[AI TRIAGE] Posted reply to {triage.tool_name} comment {triage.comment_id}", flush=True, ) except Exception as e: - print( + safe_print( f"[AI TRIAGE] Failed to post reply to comment {triage.comment_id}: {e}", flush=True, ) @@ -295,7 +297,7 @@ class GitHubOrchestrator: Returns: PRReviewResult with findings and overall assessment """ - print( + safe_print( f"[DEBUG orchestrator] review_pr() called for PR #{pr_number}", flush=True ) @@ -308,14 +310,14 @@ class GitHubOrchestrator: try: # Gather PR context - print("[DEBUG orchestrator] Creating context gatherer...", flush=True) + safe_print("[DEBUG orchestrator] Creating context gatherer...") gatherer = PRContextGatherer( self.project_dir, pr_number, repo=self.config.repo ) - print("[DEBUG orchestrator] Gathering PR context...", flush=True) + safe_print("[DEBUG orchestrator] Gathering PR context...") pr_context = await gatherer.gather() - print( + safe_print( f"[DEBUG orchestrator] Context gathered: {pr_context.title} " f"({len(pr_context.changed_files)} files, {len(pr_context.related_files)} related)", flush=True, @@ -331,14 +333,14 @@ class GitHubOrchestrator: # Allow forcing a review to bypass "already reviewed" check if should_skip and force_review and "Already reviewed" in skip_reason: - print( + safe_print( f"[BOT DETECTION] Force review requested - bypassing: {skip_reason}", flush=True, ) should_skip = False if should_skip: - print( + safe_print( f"[BOT DETECTION] Skipping PR #{pr_number}: {skip_reason}", flush=True, ) @@ -348,7 +350,7 @@ class GitHubOrchestrator: if "Already reviewed" in skip_reason: existing_review = PRReviewResult.load(self.github_dir, pr_number) if existing_review: - print( + safe_print( "[BOT DETECTION] Returning existing review (no new commits)", flush=True, ) @@ -373,14 +375,14 @@ class GitHubOrchestrator: ) # Delegate to PR Review Engine - print("[DEBUG orchestrator] Running multi-pass review...", flush=True) + safe_print("[DEBUG orchestrator] Running multi-pass review...") ( findings, structural_issues, ai_triages, quick_scan, ) = await self.pr_review_engine.run_multi_pass_review(pr_context) - print( + safe_print( f"[DEBUG orchestrator] Multi-pass review complete: " f"{len(findings)} findings, {len(structural_issues)} structural, {len(ai_triages)} AI triages", flush=True, @@ -407,12 +409,12 @@ class GitHubOrchestrator: ci_log_parts.append(f"{pending_without_awaiting} pending") if awaiting > 0: ci_log_parts.append(f"{awaiting} awaiting approval") - print( + safe_print( f"[orchestrator] CI status: {', '.join(ci_log_parts)}", flush=True, ) if awaiting > 0: - print( + safe_print( f"[orchestrator] ⚠️ {awaiting} workflow(s) from fork need maintainer approval to run", flush=True, ) @@ -426,7 +428,7 @@ class GitHubOrchestrator: has_merge_conflicts=pr_context.has_merge_conflicts, merge_state_status=pr_context.merge_state_status, ) - print( + safe_print( f"[DEBUG orchestrator] Verdict: {verdict.value} - {verdict_reasoning}", flush=True, ) @@ -471,12 +473,12 @@ class GitHubOrchestrator: blob_sha = file.get("sha", "") if filename and blob_sha: file_blobs[filename] = blob_sha - print( + safe_print( f"[Review] Captured {len(file_blobs)} file blob SHAs for follow-up tracking", flush=True, ) except Exception as e: - print( + safe_print( f"[Review] Warning: Could not capture file blobs: {e}", flush=True ) @@ -544,11 +546,11 @@ class GitHubOrchestrator: # Log full exception details for debugging error_details = f"{type(e).__name__}: {e}" full_traceback = traceback.format_exc() - print( + safe_print( f"[ERROR orchestrator] PR review failed for #{pr_number}: {error_details}", flush=True, ) - print(f"[ERROR orchestrator] Full traceback:\n{full_traceback}", flush=True) + safe_print(f"[ERROR orchestrator] Full traceback:\n{full_traceback}") result = PRReviewResult( pr_number=pr_number, @@ -577,7 +579,7 @@ class GitHubOrchestrator: Raises: ValueError: If no previous review exists for this PR """ - print( + safe_print( f"[DEBUG orchestrator] followup_review_pr() called for PR #{pr_number}", flush=True, ) @@ -622,7 +624,7 @@ class GitHubOrchestrator: # Check if context gathering failed if followup_context.error: - print( + safe_print( f"[Followup] Context gathering failed: {followup_context.error}", flush=True, ) @@ -653,7 +655,7 @@ class GitHubOrchestrator: if not has_commits and not has_file_changes: base_sha = previous_review.reviewed_commit_sha[:8] - print( + safe_print( f"[Followup] No changes since last review at {base_sha}", flush=True, ) @@ -700,7 +702,7 @@ class GitHubOrchestrator: # Use parallel orchestrator for follow-up if enabled if self.config.use_parallel_orchestrator: - print( + safe_print( "[AI] Using parallel orchestrator for follow-up review (SDK subagents)...", flush=True, ) @@ -746,7 +748,7 @@ class GitHubOrchestrator: # (CI status was already passed to AI via followup_context.ci_status) failed_checks = followup_context.ci_status.get("failed_checks", []) if failed_checks: - print( + safe_print( f"[Followup] CI checks failing: {failed_checks}", flush=True, ) @@ -1259,7 +1261,7 @@ class GitHubOrchestrator: issue["number"], result.labels_to_remove ) except Exception as e: - print(f"Failed to apply labels to #{issue['number']}: {e}") + safe_print(f"Failed to apply labels to #{issue['number']}: {e}") # Save result await result.save(self.github_dir) diff --git a/apps/backend/runners/github/runner.py b/apps/backend/runners/github/runner.py index b3934cdc..103ffc72 100644 --- a/apps/backend/runners/github/runner.py +++ b/apps/backend/runners/github/runner.py @@ -65,6 +65,11 @@ env_file = Path(__file__).parent.parent.parent / ".env" if env_file.exists(): load_dotenv(env_file) +# Initialize Sentry early to capture any startup errors +from core.sentry import capture_exception, init_sentry, set_context + +init_sentry(component="github-runner") + from debug import debug_error # Add github runner directory to path for direct imports @@ -73,6 +78,7 @@ sys.path.insert(0, str(Path(__file__).parent)) # Now import models and orchestrator directly (they use relative imports internally) from models import GitHubRunnerConfig from orchestrator import GitHubOrchestrator, ProgressCallback +from services.io_utils import safe_print def print_progress(callback: ProgressCallback) -> None: @@ -83,7 +89,7 @@ def print_progress(callback: ProgressCallback) -> None: elif callback.issue_number: prefix = f"[Issue #{callback.issue_number}] " - print(f"{prefix}[{callback.progress:3d}%] {callback.message}", flush=True) + safe_print(f"{prefix}[{callback.progress:3d}%] {callback.message}") def get_config(args) -> GitHubRunnerConfig: @@ -110,8 +116,8 @@ def get_config(args) -> GitHubRunnerConfig: break if os.environ.get("DEBUG"): - print(f"[DEBUG] gh CLI path: {gh_path}", flush=True) - print( + safe_print(f"[DEBUG] gh CLI path: {gh_path}") + safe_print( f"[DEBUG] PATH env: {os.environ.get('PATH', 'NOT SET')[:200]}...", flush=True, ) @@ -149,16 +155,20 @@ def get_config(args) -> GitHubRunnerConfig: if result.returncode == 0: repo = result.stdout.strip() elif os.environ.get("DEBUG"): - print(f"[DEBUG] gh repo view failed: {result.stderr}", flush=True) + safe_print(f"[DEBUG] gh repo view failed: {result.stderr}") except FileNotFoundError: pass # gh not installed or not in PATH if not token: - print("Error: No GitHub token found. Set GITHUB_TOKEN or run 'gh auth login'") + safe_print( + "Error: No GitHub token found. Set GITHUB_TOKEN or run 'gh auth login'" + ) sys.exit(1) if not repo: - print("Error: No GitHub repo found. Set GITHUB_REPO or run from a git repo.") + safe_print( + "Error: No GitHub repo found. Set GITHUB_REPO or run from a git repo." + ) sys.exit(1) return GitHubRunnerConfig( @@ -185,18 +195,18 @@ async def cmd_review_pr(args) -> int: debug = os.environ.get("DEBUG") if debug: - print(f"[DEBUG] Starting PR review for PR #{args.pr_number}", flush=True) - print(f"[DEBUG] Project directory: {args.project}", flush=True) - print("[DEBUG] Building config...", flush=True) + safe_print(f"[DEBUG] Starting PR review for PR #{args.pr_number}") + safe_print(f"[DEBUG] Project directory: {args.project}") + safe_print("[DEBUG] Building config...") config = get_config(args) if debug: - print( + safe_print( f"[DEBUG] Config built: repo={config.repo}, model={config.model}", flush=True, ) - print("[DEBUG] Creating orchestrator...", flush=True) + safe_print("[DEBUG] Creating orchestrator...") orchestrator = GitHubOrchestrator( project_dir=args.project, @@ -205,8 +215,8 @@ async def cmd_review_pr(args) -> int: ) if debug: - print("[DEBUG] Orchestrator created", flush=True) - print( + safe_print("[DEBUG] Orchestrator created") + safe_print( f"[DEBUG] Calling orchestrator.review_pr({args.pr_number})...", flush=True ) @@ -215,27 +225,27 @@ async def cmd_review_pr(args) -> int: result = await orchestrator.review_pr(args.pr_number, force_review=force_review) if debug: - print(f"[DEBUG] review_pr returned, success={result.success}", flush=True) + safe_print(f"[DEBUG] review_pr returned, success={result.success}") if result.success: - print(f"\n{'=' * 60}") - print(f"PR #{result.pr_number} Review Complete") - print(f"{'=' * 60}") - print(f"Status: {result.overall_status}") - print(f"Summary: {result.summary}") - print(f"Findings: {len(result.findings)}") + safe_print(f"\n{'=' * 60}") + safe_print(f"PR #{result.pr_number} Review Complete") + safe_print(f"{'=' * 60}") + safe_print(f"Status: {result.overall_status}") + safe_print(f"Summary: {result.summary}") + safe_print(f"Findings: {len(result.findings)}") if result.findings: - print("\nFindings by severity:") + safe_print("\nFindings by severity:") for f in result.findings: emoji = {"critical": "!", "high": "*", "medium": "-", "low": "."} - print( + safe_print( f" {emoji.get(f.severity.value, '?')} [{f.severity.value.upper()}] {f.title}" ) - print(f" File: {f.file}:{f.line}") + safe_print(f" File: {f.file}:{f.line}") return 0 else: - print(f"\nReview failed: {result.error}") + safe_print(f"\nReview failed: {result.error}") return 1 @@ -251,18 +261,18 @@ async def cmd_followup_review_pr(args) -> int: debug = os.environ.get("DEBUG") if debug: - print(f"[DEBUG] Starting follow-up review for PR #{args.pr_number}", flush=True) - print(f"[DEBUG] Project directory: {args.project}", flush=True) - print("[DEBUG] Building config...", flush=True) + safe_print(f"[DEBUG] Starting follow-up review for PR #{args.pr_number}") + safe_print(f"[DEBUG] Project directory: {args.project}") + safe_print("[DEBUG] Building config...") config = get_config(args) if debug: - print( + safe_print( f"[DEBUG] Config built: repo={config.repo}, model={config.model}", flush=True, ) - print("[DEBUG] Creating orchestrator...", flush=True) + safe_print("[DEBUG] Creating orchestrator...") orchestrator = GitHubOrchestrator( project_dir=args.project, @@ -271,8 +281,8 @@ async def cmd_followup_review_pr(args) -> int: ) if debug: - print("[DEBUG] Orchestrator created", flush=True) - print( + safe_print("[DEBUG] Orchestrator created") + safe_print( f"[DEBUG] Calling orchestrator.followup_review_pr({args.pr_number})...", flush=True, ) @@ -280,43 +290,43 @@ async def cmd_followup_review_pr(args) -> int: try: result = await orchestrator.followup_review_pr(args.pr_number) except ValueError as e: - print(f"\nFollow-up review failed: {e}") + safe_print(f"\nFollow-up review failed: {e}") return 1 if debug: - print( + safe_print( f"[DEBUG] followup_review_pr returned, success={result.success}", flush=True ) if result.success: - print(f"\n{'=' * 60}") - print(f"PR #{result.pr_number} Follow-up Review Complete") - print(f"{'=' * 60}") - print(f"Status: {result.overall_status}") - print(f"Is Follow-up: {result.is_followup_review}") + safe_print(f"\n{'=' * 60}") + safe_print(f"PR #{result.pr_number} Follow-up Review Complete") + safe_print(f"{'=' * 60}") + safe_print(f"Status: {result.overall_status}") + safe_print(f"Is Follow-up: {result.is_followup_review}") if result.resolved_findings: - print(f"Resolved: {len(result.resolved_findings)} finding(s)") + safe_print(f"Resolved: {len(result.resolved_findings)} finding(s)") if result.unresolved_findings: - print(f"Still Open: {len(result.unresolved_findings)} finding(s)") + safe_print(f"Still Open: {len(result.unresolved_findings)} finding(s)") if result.new_findings_since_last_review: - print( + safe_print( f"New Issues: {len(result.new_findings_since_last_review)} finding(s)" ) - print(f"\nSummary:\n{result.summary}") + safe_print(f"\nSummary:\n{result.summary}") if result.findings: - print("\nRemaining Findings:") + safe_print("\nRemaining Findings:") for f in result.findings: emoji = {"critical": "!", "high": "*", "medium": "-", "low": "."} - print( + safe_print( f" {emoji.get(f.severity.value, '?')} [{f.severity.value.upper()}] {f.title}" ) - print(f" File: {f.file}:{f.line}") + safe_print(f" File: {f.file}:{f.line}") return 0 else: - print(f"\nFollow-up review failed: {result.error}") + safe_print(f"\nFollow-up review failed: {result.error}") return 1 @@ -335,9 +345,9 @@ async def cmd_triage(args) -> int: apply_labels=args.apply_labels, ) - print(f"\n{'=' * 60}") - print(f"Triaged {len(results)} issues") - print(f"{'=' * 60}") + safe_print(f"\n{'=' * 60}") + safe_print(f"Triaged {len(results)} issues") + safe_print(f"{'=' * 60}") for r in results: flags = [] @@ -349,12 +359,12 @@ async def cmd_triage(args) -> int: flags.append("CREEP") flag_str = f" [{', '.join(flags)}]" if flags else "" - print( + safe_print( f" #{r.issue_number}: {r.category.value} (confidence: {r.confidence:.0%}){flag_str}" ) if r.labels_to_add: - print(f" + Labels: {', '.join(r.labels_to_add)}") + safe_print(f" + Labels: {', '.join(r.labels_to_add)}") return 0 @@ -371,16 +381,16 @@ async def cmd_auto_fix(args) -> int: state = await orchestrator.auto_fix_issue(args.issue_number) - print(f"\n{'=' * 60}") - print(f"Auto-Fix State for Issue #{state.issue_number}") - print(f"{'=' * 60}") - print(f"Status: {state.status.value}") + safe_print(f"\n{'=' * 60}") + safe_print(f"Auto-Fix State for Issue #{state.issue_number}") + safe_print(f"{'=' * 60}") + safe_print(f"Status: {state.status.value}") if state.spec_id: - print(f"Spec ID: {state.spec_id}") + safe_print(f"Spec ID: {state.spec_id}") if state.pr_number: - print(f"PR: #{state.pr_number}") + safe_print(f"PR: #{state.pr_number}") if state.error: - print(f"Error: {state.error}") + safe_print(f"Error: {state.error}") return 0 @@ -398,11 +408,11 @@ async def cmd_check_labels(args) -> int: issues = await orchestrator.check_auto_fix_labels() if issues: - print(f"Found {len(issues)} issues with auto-fix labels:") + safe_print(f"Found {len(issues)} issues with auto-fix labels:") for num in issues: - print(f" #{num}") + safe_print(f" #{num}") else: - print("No issues with auto-fix labels found.") + safe_print("No issues with auto-fix labels found.") return 0 @@ -419,8 +429,8 @@ async def cmd_check_new(args) -> int: issues = await orchestrator.check_new_issues() - print("JSON Output") - print(json.dumps(issues)) + safe_print("JSON Output") + safe_print(json.dumps(issues)) return 0 @@ -435,12 +445,12 @@ async def cmd_queue(args) -> int: queue = await orchestrator.get_auto_fix_queue() - print(f"\n{'=' * 60}") - print(f"Auto-Fix Queue ({len(queue)} items)") - print(f"{'=' * 60}") + safe_print(f"\n{'=' * 60}") + safe_print(f"Auto-Fix Queue ({len(queue)} items)") + safe_print(f"{'=' * 60}") if not queue: - print("Queue is empty.") + safe_print("Queue is empty.") return 0 for state in queue: @@ -455,11 +465,11 @@ async def cmd_queue(args) -> int: "failed": "ERR", } emoji = status_emoji.get(state.status.value, "???") - print(f" [{emoji}] #{state.issue_number}: {state.status.value}") + safe_print(f" [{emoji}] #{state.issue_number}: {state.status.value}") if state.pr_number: - print(f" PR: #{state.pr_number}") + safe_print(f" PR: #{state.pr_number}") if state.error: - print(f" Error: {state.error[:50]}...") + safe_print(f" Error: {state.error[:50]}...") return 0 @@ -477,22 +487,24 @@ async def cmd_batch_issues(args) -> int: issue_numbers = args.issues if args.issues else None batches = await orchestrator.batch_and_fix_issues(issue_numbers) - print(f"\n{'=' * 60}") - print(f"Created {len(batches)} batches from similar issues") - print(f"{'=' * 60}") + safe_print(f"\n{'=' * 60}") + safe_print(f"Created {len(batches)} batches from similar issues") + safe_print(f"{'=' * 60}") if not batches: - print("No batches created. Either no issues found or all issues are unique.") + safe_print( + "No batches created. Either no issues found or all issues are unique." + ) return 0 for batch in batches: issue_nums = ", ".join(f"#{i.issue_number}" for i in batch.issues) - print(f"\n Batch: {batch.batch_id}") - print(f" Issues: {issue_nums}") - print(f" Theme: {batch.theme}") - print(f" Status: {batch.status.value}") + safe_print(f"\n Batch: {batch.batch_id}") + safe_print(f" Issues: {issue_nums}") + safe_print(f" Theme: {batch.theme}") + safe_print(f" Status: {batch.status.value}") if batch.spec_id: - print(f" Spec: {batch.spec_id}") + safe_print(f" Spec: {batch.spec_id}") return 0 @@ -507,14 +519,14 @@ async def cmd_batch_status(args) -> int: status = await orchestrator.get_batch_status() - print(f"\n{'=' * 60}") - print("Batch Status") - print(f"{'=' * 60}") - print(f"Total batches: {status.get('total_batches', 0)}") - print(f"Pending: {status.get('pending', 0)}") - print(f"Processing: {status.get('processing', 0)}") - print(f"Completed: {status.get('completed', 0)}") - print(f"Failed: {status.get('failed', 0)}") + safe_print(f"\n{'=' * 60}") + safe_print("Batch Status") + safe_print(f"{'=' * 60}") + safe_print(f"Total batches: {status.get('total_batches', 0)}") + safe_print(f"Pending: {status.get('pending', 0)}") + safe_print(f"Processing: {status.get('processing', 0)}") + safe_print(f"Completed: {status.get('completed', 0)}") + safe_print(f"Failed: {status.get('failed', 0)}") return 0 @@ -543,47 +555,47 @@ async def cmd_analyze_preview(args) -> int: ) if not result.get("success"): - print(f"Error: {result.get('error', 'Unknown error')}") + safe_print(f"Error: {result.get('error', 'Unknown error')}") return 1 - print(f"\n{'=' * 60}") - print("Issue Analysis Preview") - print(f"{'=' * 60}") - print(f"Total issues: {result.get('total_issues', 0)}") - print(f"Analyzed: {result.get('analyzed_issues', 0)}") - print(f"Already batched: {result.get('already_batched', 0)}") - print(f"Proposed batches: {len(result.get('proposed_batches', []))}") - print(f"Single issues: {len(result.get('single_issues', []))}") + safe_print(f"\n{'=' * 60}") + safe_print("Issue Analysis Preview") + safe_print(f"{'=' * 60}") + safe_print(f"Total issues: {result.get('total_issues', 0)}") + safe_print(f"Analyzed: {result.get('analyzed_issues', 0)}") + safe_print(f"Already batched: {result.get('already_batched', 0)}") + safe_print(f"Proposed batches: {len(result.get('proposed_batches', []))}") + safe_print(f"Single issues: {len(result.get('single_issues', []))}") proposed_batches = result.get("proposed_batches", []) if proposed_batches: - print(f"\n{'=' * 60}") - print("Proposed Batches (for human review)") - print(f"{'=' * 60}") + safe_print(f"\n{'=' * 60}") + safe_print("Proposed Batches (for human review)") + safe_print(f"{'=' * 60}") for i, batch in enumerate(proposed_batches, 1): confidence = batch.get("confidence", 0) validated = "" if batch.get("validated") else "[NEEDS REVIEW] " - print( + safe_print( f"\n Batch {i}: {validated}{batch.get('theme', 'No theme')} ({confidence:.0%} confidence)" ) - print(f" Primary issue: #{batch.get('primary_issue')}") - print(f" Issue count: {batch.get('issue_count', 0)}") - print(f" Reasoning: {batch.get('reasoning', 'N/A')}") - print(" Issues:") + safe_print(f" Primary issue: #{batch.get('primary_issue')}") + safe_print(f" Issue count: {batch.get('issue_count', 0)}") + safe_print(f" Reasoning: {batch.get('reasoning', 'N/A')}") + safe_print(" Issues:") for item in batch.get("issues", []): similarity = item.get("similarity_to_primary", 0) - print( + safe_print( f" - #{item['issue_number']}: {item.get('title', '?')} ({similarity:.0%})" ) # Output JSON for programmatic use if getattr(args, "json", False): - print(f"\n{'=' * 60}") - print("JSON Output") - print(f"{'=' * 60}") + safe_print(f"\n{'=' * 60}") + safe_print("JSON Output") + safe_print(f"{'=' * 60}") # Print JSON on single line to avoid corruption from line-by-line stdout prefixes - print(json.dumps(result)) + safe_print(json.dumps(result)) return 0 @@ -608,24 +620,24 @@ async def cmd_approve_batches(args) -> int: with open(args.batch_file) as f: approved_batches = json.load(f) except (json.JSONDecodeError, FileNotFoundError) as e: - print(f"Error loading batch file: {e}") + safe_print(f"Error loading batch file: {e}") return 1 if not approved_batches: - print("No batches in file to approve.") + safe_print("No batches in file to approve.") return 0 - print(f"Approving and executing {len(approved_batches)} batches...") + safe_print(f"Approving and executing {len(approved_batches)} batches...") created_batches = await orchestrator.approve_and_execute_batches(approved_batches) - print(f"\n{'=' * 60}") - print(f"Created {len(created_batches)} batches") - print(f"{'=' * 60}") + safe_print(f"\n{'=' * 60}") + safe_print(f"Created {len(created_batches)} batches") + safe_print(f"{'=' * 60}") for batch in created_batches: issue_nums = ", ".join(f"#{i.issue_number}" for i in batch.issues) - print(f" {batch.batch_id}: {issue_nums}") + safe_print(f" {batch.batch_id}: {issue_nums}") return 0 @@ -800,20 +812,33 @@ def main(): handler = commands.get(args.command) if not handler: - print(f"Unknown command: {args.command}") + safe_print(f"Unknown command: {args.command}") sys.exit(1) try: + # Set context for Sentry + set_context( + "command", + { + "name": args.command, + "project": str(args.project), + "repo": args.repo or "auto-detect", + }, + ) + exit_code = asyncio.run(handler(args)) sys.exit(exit_code) except KeyboardInterrupt: - print("\nInterrupted.") + safe_print("\nInterrupted.") sys.exit(1) except Exception as e: import traceback + # Capture exception with Sentry + capture_exception(e, command=args.command) + debug_error("github_runner", "Command failed", error=str(e)) - print(f"Error: {e}") + safe_print(f"Error: {e}") traceback.print_exc() sys.exit(1) diff --git a/apps/backend/runners/github/services/batch_processor.py b/apps/backend/runners/github/services/batch_processor.py index 396a5461..5e013c4b 100644 --- a/apps/backend/runners/github/services/batch_processor.py +++ b/apps/backend/runners/github/services/batch_processor.py @@ -12,8 +12,10 @@ from pathlib import Path try: from ..models import AutoFixState, AutoFixStatus, GitHubRunnerConfig + from .io_utils import safe_print except (ImportError, ValueError, SystemError): from models import AutoFixState, AutoFixStatus, GitHubRunnerConfig + from services.io_utils import safe_print class BatchProcessor: @@ -76,10 +78,10 @@ class BatchProcessor: try: if not issues: - print("[BATCH] No issues to batch", flush=True) + safe_print("[BATCH] No issues to batch") return [] - print( + safe_print( f"[BATCH] Analyzing {len(issues)} issues for similarity...", flush=True ) @@ -123,7 +125,7 @@ class BatchProcessor: # Create batches (includes AI validation) batches = await batcher.create_batches(issues, exclude_issues) - print(f"[BATCH] Created {len(batches)} validated batches", flush=True) + safe_print(f"[BATCH] Created {len(batches)} validated batches") self._report_progress("batching", 60, f"Created {len(batches)} batches") @@ -137,7 +139,7 @@ class BatchProcessor: f"Processing batch {i + 1}/{len(batches)} ({len(issue_nums)} issues)...", ) - print( + safe_print( f"[BATCH] Batch {batch.batch_id}: {len(issue_nums)} issues - {issue_nums}", flush=True, ) @@ -164,7 +166,7 @@ class BatchProcessor: return batches except Exception as e: - print(f"[BATCH] Error batching issues: {e}", flush=True) + safe_print(f"[BATCH] Error batching issues: {e}") import traceback traceback.print_exc() @@ -204,7 +206,7 @@ class BatchProcessor: issues = issues[:max_issues] - print( + safe_print( f"[PREVIEW] Analyzing {len(issues)} issues for grouping...", flush=True ) self._report_progress("analyzing", 20, f"Analyzing {len(issues)} issues...") @@ -343,7 +345,7 @@ class BatchProcessor: reasoning = result.reasoning refined_theme = result.common_theme or refined_theme except Exception as e: - print(f"[PREVIEW] Validation error: {e}", flush=True) + safe_print(f"[PREVIEW] Validation error: {e}") validated = True confidence = 0.5 reasoning = "Validation skipped due to error" @@ -380,7 +382,7 @@ class BatchProcessor: except Exception as e: import traceback - print(f"[PREVIEW] Error: {e}", flush=True) + safe_print(f"[PREVIEW] Error: {e}") traceback.print_exc() return { "success": False, diff --git a/apps/backend/runners/github/services/followup_reviewer.py b/apps/backend/runners/github/services/followup_reviewer.py index 5c1c8bbc..a0474fe9 100644 --- a/apps/backend/runners/github/services/followup_reviewer.py +++ b/apps/backend/runners/github/services/followup_reviewer.py @@ -35,6 +35,7 @@ try: ReviewSeverity, ) from .category_utils import map_category + from .io_utils import safe_print from .prompt_manager import PromptManager from .pydantic_models import FollowupReviewResponse except (ImportError, ValueError, SystemError): @@ -47,6 +48,7 @@ except (ImportError, ValueError, SystemError): ReviewSeverity, ) from services.category_utils import map_category + from services.io_utils import safe_print from services.prompt_manager import PromptManager from services.pydantic_models import FollowupReviewResponse @@ -102,7 +104,7 @@ class FollowupReviewer: "pr_number": pr_number, } ) - print(f"[Followup] [{phase}] {message}", flush=True) + safe_print(f"[Followup] [{phase}] {message}") async def review_followup( self, @@ -691,7 +693,7 @@ Analyze this follow-up review context and provide your structured response. logger.debug( f"[Followup] Using output_format schema: {list(schema.get('properties', {}).keys())}" ) - print(f"[Followup] SDK query with output_format, model={model}", flush=True) + safe_print(f"[Followup] SDK query with output_format, model={model}") # Iterate through messages from the query # Note: max_turns=2 because structured output uses a tool call + response @@ -726,7 +728,7 @@ Analyze this follow-up review context and provide your structured response. logger.info( "[Followup] Found StructuredOutput tool use" ) - print( + safe_print( "[Followup] Using SDK structured output", flush=True, ) @@ -744,7 +746,7 @@ Analyze this follow-up review context and provide your structured response. logger.info( "[Followup] Found structured_output attribute on message" ) - print( + safe_print( "[Followup] Using SDK structured output (direct attribute)", flush=True, ) @@ -768,7 +770,7 @@ Analyze this follow-up review context and provide your structured response. except ValueError as e: # OAuth token not found logger.warning(f"No OAuth token available for AI review: {e}") - print("AI review failed: No OAuth token found", flush=True) + safe_print("AI review failed: No OAuth token found") return None except Exception as e: logger.error(f"AI review with structured output failed: {e}") diff --git a/apps/backend/runners/github/services/io_utils.py b/apps/backend/runners/github/services/io_utils.py new file mode 100644 index 00000000..d9fb4205 --- /dev/null +++ b/apps/backend/runners/github/services/io_utils.py @@ -0,0 +1,14 @@ +""" +I/O Utilities for GitHub Services +================================= + +This module re-exports safe I/O utilities from core.io_utils for +backwards compatibility. New code should import directly from core.io_utils. +""" + +from __future__ import annotations + +# Re-export from core for backwards compatibility +from core.io_utils import is_pipe_broken, reset_pipe_state, safe_print + +__all__ = ["safe_print", "is_pipe_broken", "reset_pipe_state"] diff --git a/apps/backend/runners/github/services/parallel_followup_reviewer.py b/apps/backend/runners/github/services/parallel_followup_reviewer.py index bbc23a1c..84e4c8f7 100644 --- a/apps/backend/runners/github/services/parallel_followup_reviewer.py +++ b/apps/backend/runners/github/services/parallel_followup_reviewer.py @@ -44,6 +44,7 @@ try: ReviewSeverity, ) from .category_utils import map_category + from .io_utils import safe_print from .pr_worktree_manager import PRWorktreeManager from .pydantic_models import ParallelFollowupResponse from .sdk_utils import process_sdk_stream @@ -62,6 +63,7 @@ except (ImportError, ValueError, SystemError): ) from phase_config import get_thinking_budget from services.category_utils import map_category + from services.io_utils import safe_print from services.pr_worktree_manager import PRWorktreeManager from services.pydantic_models import ParallelFollowupResponse from services.sdk_utils import process_sdk_stream @@ -456,7 +458,7 @@ The SDK will run invoked agents in parallel automatically. if head_sha and _validate_git_ref(head_sha): try: if DEBUG_MODE: - print( + safe_print( f"[Followup] DEBUG: Creating worktree for head_sha={head_sha}", flush=True, ) @@ -464,13 +466,13 @@ The SDK will run invoked agents in parallel automatically. head_sha, context.pr_number ) project_root = worktree_path - print( + safe_print( f"[Followup] Using worktree at {worktree_path.name} for PR review", flush=True, ) except Exception as e: if DEBUG_MODE: - print( + safe_print( f"[Followup] DEBUG: Worktree creation FAILED: {e}", flush=True, ) @@ -520,7 +522,7 @@ The SDK will run invoked agents in parallel automatically. async with client: await client.query(prompt) - print( + safe_print( f"[ParallelFollowup] Running orchestrator ({model})...", flush=True, ) @@ -572,7 +574,7 @@ The SDK will run invoked agents in parallel automatically. logger.info( f"[ParallelFollowup] Session complete. Agents invoked: {final_agents}" ) - print( + safe_print( f"[ParallelFollowup] Complete. Agents invoked: {final_agents}", flush=True, ) @@ -601,7 +603,7 @@ The SDK will run invoked agents in parallel automatically. "Blocked: PR has merge conflicts with base branch. " "Resolve conflicts before merge." ) - print( + safe_print( "[ParallelFollowup] ⚠️ PR has merge conflicts - blocking merge", flush=True, ) @@ -616,7 +618,7 @@ The SDK will run invoked agents in parallel automatically. ): verdict = MergeVerdict.NEEDS_REVISION verdict_reasoning = BRANCH_BEHIND_REASONING - print( + safe_print( "[ParallelFollowup] ⚠️ PR branch is behind base - needs update", flush=True, ) @@ -711,7 +713,7 @@ The SDK will run invoked agents in parallel automatically. except Exception as e: logger.error(f"[ParallelFollowup] Review failed: {e}", exc_info=True) - print(f"[ParallelFollowup] Error: {e}", flush=True) + safe_print(f"[ParallelFollowup] Error: {e}") return PRReviewResult( pr_number=context.pr_number, @@ -742,12 +744,12 @@ The SDK will run invoked agents in parallel automatically. # Log agents from structured output agents_from_output = response.agents_invoked or [] if agents_from_output: - print( + safe_print( f"[ParallelFollowup] Specialist agents invoked: {', '.join(agents_from_output)}", flush=True, ) for agent in agents_from_output: - print(f"[Agent:{agent}] Analysis complete", flush=True) + safe_print(f"[Agent:{agent}] Analysis complete") findings = [] resolved_ids = [] @@ -762,7 +764,7 @@ The SDK will run invoked agents in parallel automatically. validation_map[fv.finding_id] = fv if fv.validation_status == "dismissed_false_positive": dismissed_ids.append(fv.finding_id) - print( + safe_print( f"[ParallelFollowup] Finding {fv.finding_id} DISMISSED as false positive: {fv.explanation[:100]}", flush=True, ) @@ -774,7 +776,7 @@ The SDK will run invoked agents in parallel automatically. # Check if finding was validated and dismissed as false positive if rv.finding_id in dismissed_ids: # Finding-validator determined this was a false positive - skip it - print( + safe_print( f"[ParallelFollowup] Skipping {rv.finding_id} - dismissed as false positive by finding-validator", flush=True, ) @@ -887,27 +889,27 @@ The SDK will run invoked agents in parallel automatically. ) # Log findings summary for verification - print( + safe_print( f"[ParallelFollowup] Parsed {len(findings)} findings, " f"{len(resolved_ids)} resolved, {len(unresolved_ids)} unresolved, " f"{len(new_finding_ids)} new", flush=True, ) if dismissed_ids: - print( + safe_print( f"[ParallelFollowup] Validation: {len(dismissed_ids)} findings dismissed as false positives, " f"{confirmed_valid_count} confirmed valid, {needs_human_count} need human review", flush=True, ) if findings: - print("[ParallelFollowup] Findings summary:", flush=True) + safe_print("[ParallelFollowup] Findings summary:") for i, f in enumerate(findings, 1): validation_note = "" if f.validation_status == "confirmed_valid": validation_note = " [VALIDATED]" elif f.validation_status == "needs_human_review": validation_note = " [NEEDS HUMAN REVIEW]" - print( + safe_print( f" [{f.severity.value.upper()}] {i}. {f.title} ({f.file}:{f.line}){validation_note}", flush=True, ) diff --git a/apps/backend/runners/github/services/parallel_orchestrator_reviewer.py b/apps/backend/runners/github/services/parallel_orchestrator_reviewer.py index 254f5087..1d88a1ff 100644 --- a/apps/backend/runners/github/services/parallel_orchestrator_reviewer.py +++ b/apps/backend/runners/github/services/parallel_orchestrator_reviewer.py @@ -40,6 +40,7 @@ try: ReviewSeverity, ) from .category_utils import map_category + from .io_utils import safe_print from .pr_worktree_manager import PRWorktreeManager from .pydantic_models import ParallelOrchestratorResponse from .sdk_utils import process_sdk_stream @@ -58,6 +59,7 @@ except (ImportError, ValueError, SystemError): ) from phase_config import get_thinking_budget from services.category_utils import map_category + from services.io_utils import safe_print from services.pr_worktree_manager import PRWorktreeManager from services.pydantic_models import ParallelOrchestratorResponse from services.sdk_utils import process_sdk_stream @@ -380,12 +382,12 @@ The SDK will run invoked agents in parallel automatically. agents: List of agent names that were invoked """ if agents: - print( + safe_print( f"[ParallelOrchestrator] Specialist agents invoked: {', '.join(agents)}", flush=True, ) for agent in agents: - print(f"[Agent:{agent}] Analysis complete", flush=True) + safe_print(f"[Agent:{agent}] Analysis complete") def _log_findings_summary(self, findings: list[PRReviewFinding]) -> None: """Log findings summary for verification. @@ -394,13 +396,13 @@ The SDK will run invoked agents in parallel automatically. findings: List of findings to summarize """ if findings: - print( + safe_print( f"[ParallelOrchestrator] Parsed {len(findings)} findings from structured output", flush=True, ) - print("[ParallelOrchestrator] Findings summary:", flush=True) + safe_print("[ParallelOrchestrator] Findings summary:") for i, f in enumerate(findings, 1): - print( + safe_print( f" [{f.severity.value.upper()}] {i}. {f.title} ({f.file}:{f.line})", flush=True, ) @@ -474,15 +476,15 @@ The SDK will run invoked agents in parallel automatically. head_sha = context.head_sha or context.head_branch if DEBUG_MODE: - print( + safe_print( f"[PRReview] DEBUG: context.head_sha='{context.head_sha}'", flush=True, ) - print( + safe_print( f"[PRReview] DEBUG: context.head_branch='{context.head_branch}'", flush=True, ) - print(f"[PRReview] DEBUG: resolved head_sha='{head_sha}'", flush=True) + safe_print(f"[PRReview] DEBUG: resolved head_sha='{head_sha}'") # SECURITY: Validate the resolved head_sha (whether SHA or branch name) # This catches invalid refs early before subprocess calls @@ -495,7 +497,7 @@ The SDK will run invoked agents in parallel automatically. if not head_sha: if DEBUG_MODE: - print("[PRReview] DEBUG: No head_sha - using fallback", flush=True) + safe_print("[PRReview] DEBUG: No head_sha - using fallback") logger.warning( "[ParallelOrchestrator] No head_sha available, using current checkout" ) @@ -507,7 +509,7 @@ The SDK will run invoked agents in parallel automatically. ) else: if DEBUG_MODE: - print( + safe_print( f"[PRReview] DEBUG: Creating worktree for head_sha={head_sha}", flush=True, ) @@ -517,14 +519,14 @@ The SDK will run invoked agents in parallel automatically. ) project_root = worktree_path if DEBUG_MODE: - print( + safe_print( f"[PRReview] DEBUG: Using worktree as " f"project_root={project_root}", flush=True, ) except (RuntimeError, ValueError) as e: if DEBUG_MODE: - print( + safe_print( f"[PRReview] DEBUG: Worktree creation FAILED: {e}", flush=True, ) @@ -564,7 +566,7 @@ The SDK will run invoked agents in parallel automatically. async with client: await client.query(prompt) - print( + safe_print( f"[ParallelOrchestrator] Running orchestrator ({model})...", flush=True, ) @@ -608,7 +610,7 @@ The SDK will run invoked agents in parallel automatically. logger.info( f"[ParallelOrchestrator] Session complete. Agents invoked: {final_agents}" ) - print( + safe_print( f"[ParallelOrchestrator] Complete. Agents invoked: {final_agents}", flush=True, ) diff --git a/apps/backend/runners/github/services/pr_review_engine.py b/apps/backend/runners/github/services/pr_review_engine.py index d8832539..f68a8ec3 100644 --- a/apps/backend/runners/github/services/pr_review_engine.py +++ b/apps/backend/runners/github/services/pr_review_engine.py @@ -21,6 +21,7 @@ try: ReviewPass, StructuralIssue, ) + from .io_utils import safe_print from .prompt_manager import PromptManager from .response_parsers import ResponseParser except (ImportError, ValueError, SystemError): @@ -32,6 +33,7 @@ except (ImportError, ValueError, SystemError): ReviewPass, StructuralIssue, ) + from services.io_utils import safe_print from services.prompt_manager import PromptManager from services.response_parsers import ResponseParser @@ -80,19 +82,19 @@ class PRReviewEngine: total_changes = context.total_additions + context.total_deletions if total_changes > 200: - print( + safe_print( f"[AI] Deep analysis needed: {total_changes} lines changed", flush=True ) return True complexity = scan_result.get("complexity", "low") if complexity in ["high", "medium"]: - print(f"[AI] Deep analysis needed: {complexity} complexity", flush=True) + safe_print(f"[AI] Deep analysis needed: {complexity} complexity") return True risk_areas = scan_result.get("risk_areas", []) if risk_areas: - print( + safe_print( f"[AI] Deep analysis needed: {len(risk_areas)} risk areas", flush=True ) return True @@ -111,7 +113,7 @@ class PRReviewEngine: seen.add(key) unique.append(f) else: - print( + safe_print( f"[AI] Skipping duplicate finding: {f.file}:{f.line} - {f.title}", flush=True, ) @@ -171,7 +173,7 @@ class PRReviewEngine: # If diff is empty/truncated, build composite from individual file patches if context.diff_truncated or not context.diff: - print( + safe_print( f"[AI] Building composite diff from {len(context.changed_files)} file patches...", flush=True, ) @@ -260,7 +262,7 @@ class PRReviewEngine: error_msg = f"Review pass {review_pass.value} failed: {e}" logger.error(error_msg) logger.error(f"Traceback: {traceback.format_exc()}") - print(f"[AI] ERROR: {error_msg}", flush=True) + safe_print(f"[AI] ERROR: {error_msg}") # Re-raise to allow caller to handle or track partial failures raise RuntimeError(error_msg) from e @@ -281,7 +283,7 @@ class PRReviewEngine: """ # Use parallel orchestrator with SDK subagents if enabled if self.config.use_parallel_orchestrator: - print( + safe_print( "[AI] Using parallel orchestrator PR review (SDK subagents)...", flush=True, ) @@ -303,7 +305,7 @@ class PRReviewEngine: result = await orchestrator.review(context) - print( + safe_print( f"[PR Review Engine] Parallel orchestrator returned {len(result.findings)} findings", flush=True, ) @@ -322,7 +324,7 @@ class PRReviewEngine: ai_triages = [] # Pass 1: Quick Scan (must run first - determines if deep analysis needed) - print("[AI] Pass 1/6: Quick Scan - Understanding scope...", flush=True) + safe_print("[AI] Pass 1/6: Quick Scan - Understanding scope...") self._report_progress( "analyzing", 35, @@ -339,7 +341,7 @@ class PRReviewEngine: parallel_tasks = [] task_names = [] - print("[AI] Running passes 2-6 in parallel...", flush=True) + safe_print("[AI] Running passes 2-6 in parallel...") self._report_progress( "analyzing", 50, @@ -348,50 +350,50 @@ class PRReviewEngine: ) async def run_security_pass(): - print( + safe_print( "[AI] Pass 2/6: Security Review - Analyzing vulnerabilities...", flush=True, ) findings = await self.run_review_pass(ReviewPass.SECURITY, context) - print(f"[AI] Security pass complete: {len(findings)} findings", flush=True) + safe_print(f"[AI] Security pass complete: {len(findings)} findings") return ("security", findings) async def run_quality_pass(): - print( + safe_print( "[AI] Pass 3/6: Quality Review - Checking code quality...", flush=True ) findings = await self.run_review_pass(ReviewPass.QUALITY, context) - print(f"[AI] Quality pass complete: {len(findings)} findings", flush=True) + safe_print(f"[AI] Quality pass complete: {len(findings)} findings") return ("quality", findings) async def run_structural_pass(): - print( + safe_print( "[AI] Pass 4/6: Structural Review - Checking for feature creep...", flush=True, ) result_text = await self._run_structural_pass(context) issues = self.parser.parse_structural_issues(result_text) - print(f"[AI] Structural pass complete: {len(issues)} issues", flush=True) + safe_print(f"[AI] Structural pass complete: {len(issues)} issues") return ("structural", issues) async def run_ai_triage_pass(): - print( + safe_print( "[AI] Pass 5/6: AI Comment Triage - Verifying other AI comments...", flush=True, ) result_text = await self._run_ai_triage_pass(context) triages = self.parser.parse_ai_comment_triages(result_text) - print( + safe_print( f"[AI] AI triage complete: {len(triages)} comments triaged", flush=True ) return ("ai_triage", triages) async def run_deep_pass(): - print( + safe_print( "[AI] Pass 6/6: Deep Analysis - Reviewing business logic...", flush=True ) findings = await self.run_review_pass(ReviewPass.DEEP_ANALYSIS, context) - print(f"[AI] Deep analysis complete: {len(findings)} findings", flush=True) + safe_print(f"[AI] Deep analysis complete: {len(findings)} findings") return ("deep", findings) # Always run security, quality, structural @@ -408,22 +410,22 @@ class PRReviewEngine: if has_ai_comments: parallel_tasks.append(run_ai_triage_pass()) task_names.append("AI Triage") - print( + safe_print( f"[AI] Found {len(context.ai_bot_comments)} AI comments to triage", flush=True, ) else: - print("[AI] Pass 5/6: Skipped (no AI comments to triage)", flush=True) + safe_print("[AI] Pass 5/6: Skipped (no AI comments to triage)") # Only run deep analysis if needed if needs_deep: parallel_tasks.append(run_deep_pass()) task_names.append("Deep Analysis") else: - print("[AI] Pass 6/6: Skipped (changes not complex enough)", flush=True) + safe_print("[AI] Pass 6/6: Skipped (changes not complex enough)") # Run all passes in parallel - print( + safe_print( f"[AI] Executing {len(parallel_tasks)} passes in parallel: {', '.join(task_names)}", flush=True, ) @@ -432,7 +434,7 @@ class PRReviewEngine: # Collect results from all parallel passes for i, result in enumerate(results): if isinstance(result, Exception): - print(f"[AI] Pass '{task_names[i]}' failed: {result}", flush=True) + safe_print(f"[AI] Pass '{task_names[i]}' failed: {result}") elif isinstance(result, tuple): pass_type, data = result if pass_type in ("security", "quality", "deep"): @@ -450,12 +452,12 @@ class PRReviewEngine: ) # Deduplicate findings - print( + safe_print( f"[AI] Deduplicating {len(all_findings)} findings from all passes...", flush=True, ) unique_findings = self.deduplicate_findings(all_findings) - print( + safe_print( f"[AI] Multi-pass review complete: {len(unique_findings)} findings, " f"{len(structural_issues)} structural issues, {len(ai_triages)} AI triages", flush=True, @@ -509,7 +511,7 @@ class PRReviewEngine: if block_type == "TextBlock" and hasattr(block, "text"): result_text += block.text except Exception as e: - print(f"[AI] Structural pass error: {e}", flush=True) + safe_print(f"[AI] Structural pass error: {e}") return result_text @@ -567,7 +569,7 @@ class PRReviewEngine: if block_type == "TextBlock" and hasattr(block, "text"): result_text += block.text except Exception as e: - print(f"[AI] AI triage pass error: {e}", flush=True) + safe_print(f"[AI] AI triage pass error: {e}") return result_text diff --git a/apps/backend/runners/github/services/response_parsers.py b/apps/backend/runners/github/services/response_parsers.py index 2df83ea0..c0b31e87 100644 --- a/apps/backend/runners/github/services/response_parsers.py +++ b/apps/backend/runners/github/services/response_parsers.py @@ -21,6 +21,7 @@ try: TriageCategory, TriageResult, ) + from .io_utils import safe_print except (ImportError, ValueError, SystemError): from models import ( AICommentTriage, @@ -32,6 +33,7 @@ except (ImportError, ValueError, SystemError): TriageCategory, TriageResult, ) + from services.io_utils import safe_print # Evidence-based validation replaces confidence scoring # Findings without evidence are filtered out instead of using confidence thresholds @@ -57,10 +59,10 @@ class ResponseParser: ) if json_match: result = json.loads(json_match.group(1)) - print(f"[AI] Quick scan result: {result}", flush=True) + safe_print(f"[AI] Quick scan result: {result}") return result except (json.JSONDecodeError, ValueError) as e: - print(f"[AI] Failed to parse scan result: {e}", flush=True) + safe_print(f"[AI] Failed to parse scan result: {e}") return default_result @@ -87,7 +89,7 @@ class ResponseParser: # Apply evidence-based validation if require_evidence and len(evidence.strip()) < MIN_EVIDENCE_LENGTH: - print( + safe_print( f"[AI] Dropped finding '{f.get('title', 'unknown')}': " f"insufficient evidence ({len(evidence.strip())} chars < {MIN_EVIDENCE_LENGTH})", flush=True, @@ -117,7 +119,7 @@ class ResponseParser: ) ) except (json.JSONDecodeError, KeyError, ValueError) as e: - print(f"Failed to parse findings: {e}") + safe_print(f"Failed to parse findings: {e}") return findings @@ -147,7 +149,7 @@ class ResponseParser: ) ) except (json.JSONDecodeError, KeyError, ValueError) as e: - print(f"Failed to parse structural issues: {e}") + safe_print(f"Failed to parse structural issues: {e}") return issues @@ -180,7 +182,7 @@ class ResponseParser: ) ) except (json.JSONDecodeError, KeyError, ValueError) as e: - print(f"Failed to parse AI comment triages: {e}") + safe_print(f"Failed to parse AI comment triages: {e}") return triages @@ -218,6 +220,6 @@ class ResponseParser: result.comment = data.get("comment") except (json.JSONDecodeError, KeyError, ValueError) as e: - print(f"Failed to parse triage result: {e}") + safe_print(f"Failed to parse triage result: {e}") return result diff --git a/apps/backend/runners/github/services/sdk_utils.py b/apps/backend/runners/github/services/sdk_utils.py index 7471f163..c6ac7d89 100644 --- a/apps/backend/runners/github/services/sdk_utils.py +++ b/apps/backend/runners/github/services/sdk_utils.py @@ -15,6 +15,11 @@ import os from collections.abc import Callable from typing import Any +try: + from .io_utils import safe_print +except (ImportError, ValueError, SystemError): + from core.io_utils import safe_print + logger = logging.getLogger(__name__) # Check if debug mode is enabled @@ -66,9 +71,9 @@ async def process_sdk_stream( # Track subagent tool IDs to log their results subagent_tool_ids: dict[str, str] = {} # tool_id -> agent_name - print(f"[{context_name}] Processing SDK stream...", flush=True) + safe_print(f"[{context_name}] Processing SDK stream...") if DEBUG_MODE: - print(f"[DEBUG {context_name}] Awaiting response stream...", flush=True) + safe_print(f"[DEBUG {context_name}] Awaiting response stream...") try: async for msg in client.receive_response(): @@ -81,9 +86,8 @@ async def process_sdk_stream( msg_details = "" if hasattr(msg, "type"): msg_details = f" (type={msg.type})" - print( - f"[DEBUG {context_name}] Message #{msg_count}: {msg_type}{msg_details}", - flush=True, + safe_print( + f"[DEBUG {context_name}] Message #{msg_count}: {msg_type}{msg_details}" ) # Track thinking blocks @@ -94,16 +98,14 @@ async def process_sdk_stream( msg, "text", "" ) if thinking_text: - print( - f"[{context_name}] AI thinking: {len(thinking_text)} chars", - flush=True, + safe_print( + f"[{context_name}] AI thinking: {len(thinking_text)} chars" ) if DEBUG_MODE: # Show first 200 chars of thinking preview = thinking_text[:200].replace("\n", " ") - print( - f"[DEBUG {context_name}] Thinking preview: {preview}...", - flush=True, + safe_print( + f"[DEBUG {context_name}] Thinking preview: {preview}..." ) # Invoke callback if on_thinking: @@ -118,9 +120,8 @@ async def process_sdk_stream( tool_input = getattr(msg, "input", {}) if DEBUG_MODE: - print( - f"[DEBUG {context_name}] Tool call: {tool_name} (id={tool_id})", - flush=True, + safe_print( + f"[DEBUG {context_name}] Tool call: {tool_name} (id={tool_id})" ) if tool_name == "Task": @@ -129,9 +130,7 @@ async def process_sdk_stream( agents_invoked.append(agent_name) # Track this tool ID to log its result later subagent_tool_ids[tool_id] = agent_name - print( - f"[{context_name}] Invoked agent: {agent_name}", flush=True - ) + safe_print(f"[{context_name}] Invoked agent: {agent_name}") elif tool_name == "StructuredOutput": if tool_input: # Warn if overwriting existing structured output @@ -141,19 +140,13 @@ async def process_sdk_stream( f"overwriting previous output" ) structured_output = tool_input - print( - f"[{context_name}] Received structured output", - flush=True, - ) + safe_print(f"[{context_name}] Received structured output") # Invoke callback if on_structured_output: on_structured_output(tool_input) elif DEBUG_MODE: # Log other tool calls in debug mode - print( - f"[DEBUG {context_name}] Other tool: {tool_name}", - flush=True, - ) + safe_print(f"[DEBUG {context_name}] Other tool: {tool_name}") # Invoke callback for all tool uses if on_tool_use: @@ -180,15 +173,13 @@ async def process_sdk_stream( result_preview = ( str(result_content)[:600].replace("\n", " ").strip() ) - print( - f"[Agent:{agent_name}] {status}: {result_preview}{'...' if len(str(result_content)) > 600 else ''}", - flush=True, + safe_print( + f"[Agent:{agent_name}] {status}: {result_preview}{'...' if len(str(result_content)) > 600 else ''}" ) elif DEBUG_MODE: status = "ERROR" if is_error else "OK" - print( - f"[DEBUG {context_name}] Tool result: {tool_id} [{status}]", - flush=True, + safe_print( + f"[DEBUG {context_name}] Tool result: {tool_id} [{status}]" ) # Invoke callback @@ -214,9 +205,8 @@ async def process_sdk_stream( if agent_name not in agents_invoked: agents_invoked.append(agent_name) subagent_tool_ids[tool_id] = agent_name - print( - f"[{context_name}] Invoking agent: {agent_name}", - flush=True, + safe_print( + f"[{context_name}] Invoking agent: {agent_name}" ) elif tool_name == "StructuredOutput": if tool_input: @@ -242,9 +232,8 @@ async def process_sdk_stream( # Always print text content preview (not just in DEBUG_MODE) text_preview = block.text[:500].replace("\n", " ").strip() if text_preview: - print( - f"[{context_name}] AI response: {text_preview}{'...' if len(block.text) > 500 else ''}", - flush=True, + safe_print( + f"[{context_name}] AI response: {text_preview}{'...' if len(block.text) > 500 else ''}" ) # Invoke callback if on_text: @@ -304,9 +293,8 @@ async def process_sdk_stream( result_preview = ( str(result_content)[:600].replace("\n", " ").strip() ) - print( - f"[Agent:{agent_name}] {status}: {result_preview}{'...' if len(str(result_content)) > 600 else ''}", - flush=True, + safe_print( + f"[Agent:{agent_name}] {status}: {result_preview}{'...' if len(str(result_content)) > 600 else ''}" ) # Invoke callback @@ -319,25 +307,25 @@ async def process_sdk_stream( f"[{context_name}] Error processing message #{msg_count}: {msg_error}" ) if DEBUG_MODE: - print( - f"[DEBUG {context_name}] Message processing error: {msg_error}", - flush=True, + safe_print( + f"[DEBUG {context_name}] Message processing error: {msg_error}" ) # Continue processing subsequent messages + except BrokenPipeError: + # Pipe closed by parent process - expected during shutdown + stream_error = "Output pipe closed" + logger.debug(f"[{context_name}] Output pipe closed by parent process") except Exception as e: # Log stream-level errors stream_error = str(e) logger.error(f"[{context_name}] SDK stream processing failed: {e}") - print(f"[{context_name}] ERROR: Stream processing failed: {e}", flush=True) + safe_print(f"[{context_name}] ERROR: Stream processing failed: {e}") if DEBUG_MODE: - print( - f"[DEBUG {context_name}] Session ended. Total messages: {msg_count}", - flush=True, - ) + safe_print(f"[DEBUG {context_name}] Session ended. Total messages: {msg_count}") - print(f"[{context_name}] Session ended. Total messages: {msg_count}", flush=True) + safe_print(f"[{context_name}] Session ended. Total messages: {msg_count}") return { "result_text": result_text, diff --git a/apps/backend/runners/gitlab/orchestrator.py b/apps/backend/runners/gitlab/orchestrator.py index bc33aa20..088ecca8 100644 --- a/apps/backend/runners/gitlab/orchestrator.py +++ b/apps/backend/runners/gitlab/orchestrator.py @@ -36,6 +36,17 @@ except ImportError: ) from services import MRReviewEngine +# Import safe_print for BrokenPipeError handling +try: + from core.io_utils import safe_print +except ImportError: + # Fallback for direct script execution + import sys + from pathlib import Path + + sys.path.insert(0, str(Path(__file__).parent.parent.parent)) + from core.io_utils import safe_print + @dataclass class ProgressCallback: @@ -121,7 +132,7 @@ class GitLabOrchestrator: async def _gather_mr_context(self, mr_iid: int) -> MRContext: """Gather context for an MR.""" - print(f"[GitLab] Fetching MR !{mr_iid} data...", flush=True) + safe_print(f"[GitLab] Fetching MR !{mr_iid} data...") # Get MR details mr_data = self.client.get_mr(mr_iid) @@ -187,7 +198,7 @@ class GitLabOrchestrator: Returns: MRReviewResult with findings and overall assessment """ - print(f"[GitLab] Starting review for MR !{mr_iid}", flush=True) + safe_print(f"[GitLab] Starting review for MR !{mr_iid}") self._report_progress( "gathering_context", @@ -199,10 +210,9 @@ class GitLabOrchestrator: try: # Gather MR context context = await self._gather_mr_context(mr_iid) - print( + safe_print( f"[GitLab] Context gathered: {context.title} " - f"({len(context.changed_files)} files, {context.total_additions}+/{context.total_deletions}-)", - flush=True, + f"({len(context.changed_files)} files, {context.total_additions}+/{context.total_deletions}-)" ) self._report_progress( @@ -213,7 +223,7 @@ class GitLabOrchestrator: findings, verdict, summary, blockers = await self.review_engine.run_review( context ) - print(f"[GitLab] Review complete: {len(findings)} findings", flush=True) + safe_print(f"[GitLab] Review complete: {len(findings)} findings") # Map verdict to overall_status if verdict == MergeVerdict.BLOCKED: @@ -264,7 +274,7 @@ class GitLabOrchestrator: error_msg = f"MR !{mr_iid} not found in GitLab." elif e.code == 429: error_msg = "GitLab rate limit exceeded. Please try again later." - print(f"[GitLab] Review failed for !{mr_iid}: {error_msg}", flush=True) + safe_print(f"[GitLab] Review failed for !{mr_iid}: {error_msg}") result = MRReviewResult( mr_iid=mr_iid, project=self.config.project, @@ -276,7 +286,7 @@ class GitLabOrchestrator: except json.JSONDecodeError as e: error_msg = f"Invalid JSON response from GitLab: {e}" - print(f"[GitLab] Review failed for !{mr_iid}: {error_msg}", flush=True) + safe_print(f"[GitLab] Review failed for !{mr_iid}: {error_msg}") result = MRReviewResult( mr_iid=mr_iid, project=self.config.project, @@ -288,7 +298,7 @@ class GitLabOrchestrator: except OSError as e: error_msg = f"File system error: {e}" - print(f"[GitLab] Review failed for !{mr_iid}: {error_msg}", flush=True) + safe_print(f"[GitLab] Review failed for !{mr_iid}: {error_msg}") result = MRReviewResult( mr_iid=mr_iid, project=self.config.project, @@ -302,8 +312,8 @@ class GitLabOrchestrator: # Catch-all for unexpected errors, with full traceback for debugging error_details = f"{type(e).__name__}: {e}" full_traceback = traceback.format_exc() - print(f"[GitLab] Review failed for !{mr_iid}: {error_details}", flush=True) - print(f"[GitLab] Traceback:\n{full_traceback}", flush=True) + safe_print(f"[GitLab] Review failed for !{mr_iid}: {error_details}") + safe_print(f"[GitLab] Traceback:\n{full_traceback}") result = MRReviewResult( mr_iid=mr_iid, @@ -326,7 +336,7 @@ class GitLabOrchestrator: Returns: MRReviewResult with follow-up analysis """ - print(f"[GitLab] Starting follow-up review for MR !{mr_iid}", flush=True) + safe_print(f"[GitLab] Starting follow-up review for MR !{mr_iid}") # Load previous review previous_review = MRReviewResult.load(self.gitlab_dir, mr_iid) diff --git a/apps/backend/runners/gitlab/runner.py b/apps/backend/runners/gitlab/runner.py index d4f61827..a24b284c 100644 --- a/apps/backend/runners/gitlab/runner.py +++ b/apps/backend/runners/gitlab/runner.py @@ -38,6 +38,7 @@ if env_file.exists(): # Add gitlab runner directory to path for direct imports sys.path.insert(0, str(Path(__file__).parent)) +from core.io_utils import safe_print from models import GitLabRunnerConfig from orchestrator import GitLabOrchestrator, ProgressCallback @@ -48,7 +49,7 @@ def print_progress(callback: ProgressCallback) -> None: if callback.mr_iid: prefix = f"[MR !{callback.mr_iid}] " - print(f"{prefix}[{callback.progress:3d}%] {callback.message}", flush=True) + safe_print(f"{prefix}[{callback.progress:3d}%] {callback.message}") def get_config(args) -> GitLabRunnerConfig: @@ -122,27 +123,24 @@ async def cmd_review_mr(args) -> int: sys.stdout.reconfigure(line_buffering=True) sys.stderr.reconfigure(line_buffering=True) - print(f"[DEBUG] Starting MR review for MR !{args.mr_iid}", flush=True) - print(f"[DEBUG] Project directory: {args.project_dir}", flush=True) + safe_print(f"[DEBUG] Starting MR review for MR !{args.mr_iid}") + safe_print(f"[DEBUG] Project directory: {args.project_dir}") - print("[DEBUG] Building config...", flush=True) + safe_print("[DEBUG] Building config...") config = get_config(args) - print( - f"[DEBUG] Config built: project={config.project}, model={config.model}", - flush=True, - ) + safe_print(f"[DEBUG] Config built: project={config.project}, model={config.model}") - print("[DEBUG] Creating orchestrator...", flush=True) + safe_print("[DEBUG] Creating orchestrator...") orchestrator = GitLabOrchestrator( project_dir=args.project_dir, config=config, progress_callback=print_progress, ) - print("[DEBUG] Orchestrator created", flush=True) + safe_print("[DEBUG] Orchestrator created") - print(f"[DEBUG] Calling orchestrator.review_mr({args.mr_iid})...", flush=True) + safe_print(f"[DEBUG] Calling orchestrator.review_mr({args.mr_iid})...") result = await orchestrator.review_mr(args.mr_iid) - print(f"[DEBUG] review_mr returned, success={result.success}", flush=True) + safe_print(f"[DEBUG] review_mr returned, success={result.success}") if result.success: print(f"\n{'=' * 60}") @@ -174,27 +172,22 @@ async def cmd_followup_review_mr(args) -> int: sys.stdout.reconfigure(line_buffering=True) sys.stderr.reconfigure(line_buffering=True) - print(f"[DEBUG] Starting follow-up review for MR !{args.mr_iid}", flush=True) - print(f"[DEBUG] Project directory: {args.project_dir}", flush=True) + safe_print(f"[DEBUG] Starting follow-up review for MR !{args.mr_iid}") + safe_print(f"[DEBUG] Project directory: {args.project_dir}") - print("[DEBUG] Building config...", flush=True) + safe_print("[DEBUG] Building config...") config = get_config(args) - print( - f"[DEBUG] Config built: project={config.project}, model={config.model}", - flush=True, - ) + safe_print(f"[DEBUG] Config built: project={config.project}, model={config.model}") - print("[DEBUG] Creating orchestrator...", flush=True) + safe_print("[DEBUG] Creating orchestrator...") orchestrator = GitLabOrchestrator( project_dir=args.project_dir, config=config, progress_callback=print_progress, ) - print("[DEBUG] Orchestrator created", flush=True) + safe_print("[DEBUG] Orchestrator created") - print( - f"[DEBUG] Calling orchestrator.followup_review_mr({args.mr_iid})...", flush=True - ) + safe_print(f"[DEBUG] Calling orchestrator.followup_review_mr({args.mr_iid})...") try: result = await orchestrator.followup_review_mr(args.mr_iid) @@ -202,7 +195,7 @@ async def cmd_followup_review_mr(args) -> int: print(f"\nFollow-up review failed: {e}") return 1 - print(f"[DEBUG] followup_review_mr returned, success={result.success}", flush=True) + safe_print(f"[DEBUG] followup_review_mr returned, success={result.success}") if result.success: print(f"\n{'=' * 60}") diff --git a/apps/backend/runners/gitlab/services/mr_review_engine.py b/apps/backend/runners/gitlab/services/mr_review_engine.py index ef8ef9aa..0a8d518a 100644 --- a/apps/backend/runners/gitlab/services/mr_review_engine.py +++ b/apps/backend/runners/gitlab/services/mr_review_engine.py @@ -34,6 +34,17 @@ except ImportError: ReviewSeverity, ) +# Import safe_print for BrokenPipeError handling +try: + from core.io_utils import safe_print +except ImportError: + # Fallback for direct script execution + import sys + from pathlib import Path as PathLib + + sys.path.insert(0, str(PathLib(__file__).parent.parent.parent.parent)) + from core.io_utils import safe_print + @dataclass class ProgressCallback: @@ -246,7 +257,7 @@ Provide your review in the following JSON format: return self._parse_review_result(result_text) except Exception as e: - print(f"[AI] Review error: {e}", flush=True) + safe_print(f"[AI] Review error: {e}") raise RuntimeError(f"Review failed: {e}") from e def _parse_review_result( @@ -297,14 +308,11 @@ Provide your review in the following JSON format: f"{finding.title} ({finding.file}:{finding.line})" ) except (ValueError, KeyError) as e: - print(f"[AI] Skipping invalid finding: {e}", flush=True) + safe_print(f"[AI] Skipping invalid finding: {e}") except json.JSONDecodeError as e: - print(f"[AI] Failed to parse JSON: {e}", flush=True) - print( - f"[AI] Raw response (first 500 chars): {result_text[:500]}", - flush=True, - ) + safe_print(f"[AI] Failed to parse JSON: {e}") + safe_print(f"[AI] Raw response (first 500 chars): {result_text[:500]}") summary = "Review completed but failed to parse structured output. Please re-run the review." # Return with empty findings but keep verdict as READY_TO_MERGE # since we couldn't determine if there are actual issues diff --git a/apps/backend/runners/spec_runner.py b/apps/backend/runners/spec_runner.py index 30adbf3f..39868daf 100644 --- a/apps/backend/runners/spec_runner.py +++ b/apps/backend/runners/spec_runner.py @@ -93,6 +93,11 @@ if env_file.exists(): elif dev_env_file.exists(): load_dotenv(dev_env_file) +# Initialize Sentry early to capture any startup errors +from core.sentry import capture_exception, init_sentry + +init_sentry(component="spec-runner") + from debug import debug, debug_error, debug_section, debug_success from phase_config import resolve_model_id from review import ReviewState @@ -370,6 +375,14 @@ Examples: f"To continue: python auto-claude/spec_runner.py --continue {orchestrator.spec_dir.name}" ) sys.exit(1) + except Exception as e: + # Capture unexpected errors to Sentry + capture_exception( + e, spec_dir=str(orchestrator.spec_dir) if orchestrator else None + ) + debug_error("spec_runner", f"Unexpected error: {e}") + print(f"\n\nUnexpected error: {e}") + sys.exit(1) if __name__ == "__main__":