fix(backend): add Sentry integration and fix broken pipe errors (#991)

* fix(backend): add Sentry integration and fix broken pipe errors

- Add sentry-sdk to Python backend for error tracking
- Create safe_print() utility to handle BrokenPipeError gracefully
- Initialize Sentry in CLI, GitHub runner, and spec runner entry points
- Use same SENTRY_DSN environment variable as Electron frontend
- Apply privacy path masking (usernames removed from stack traces)

Fixes "Review Failed: [Errno 32] Broken pipe" error in PR review

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

* fix(backend): address PR review findings for Sentry integration

- Fix ruff linting errors (unused imports, import sorting)
- Add path masking to set_context() and set_tag() for privacy
- Add defensive path masking to capture_exception() kwargs
- Add debug logging for bare except clauses in sentry.py
- Add top-level error handler in cli/main.py with Sentry capture
- Add error handling with Sentry capture in spec_runner.py
- Move safe_print to core/io_utils.py for broader reuse
- Migrate GitLab runner files to use safe_print()
- Add fallback import pattern in sdk_utils.py

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

* style: apply ruff formatting

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

* fix(backend): address CodeRabbit review findings for Sentry and io_utils

- Add path masking to capture_message() kwargs for privacy consistency
- Add recursion depth limit (50) to _mask_object_paths() to prevent stack overflow
- Add WSL path masking support (/mnt/[a-z]/Users/...)
- Add consistent ImportError debug logging across Sentry wrapper functions
- Add ValueError handling in safe_print() for closed stdout scenarios
- Improve reset_pipe_state() documentation with usage warnings

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 4.5 <noreply@anthropic.com>

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

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

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

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

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,

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)

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)
 

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,

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}")

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"]

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,
                     )

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,
             )

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
 

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

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,

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)

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}")

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

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__":