refactor: break up capture_error() into testable helpers (#523)

Co-authored-by: Kimi Agent <kimi@timmy.local>
Co-committed-by: Kimi Agent <kimi@timmy.local>

src/infrastructure/error_capture.py

@@ -100,36 +100,14 @@ def _get_git_context() -> dict:
         return {"branch": "unknown", "commit": "unknown"}
 
 
-def capture_error(
-    exc: Exception,
-    source: str = "unknown",
-    context: dict | None = None,
-) -> str | None:
-    """Capture an error and optionally create a bug report.
-
-    Args:
-        exc: The exception to capture
-        source: Module/component where the error occurred
-        context: Optional dict of extra context (request path, etc.)
+def _extract_traceback_info(exc: Exception) -> tuple[str, str, int]:
+    """Extract formatted traceback, affected file, and line number.
 
     Returns:
-        Task ID of the created bug report, or None if deduplicated/disabled
+        Tuple of (traceback_string, affected_file, affected_line).
     """
-    from config import settings
-
-    if not settings.error_feedback_enabled:
-        return None
-
-    error_hash = _stack_hash(exc)
-
-    if _is_duplicate(error_hash):
-        logger.debug("Duplicate error suppressed: %s (hash=%s)", exc, error_hash)
-        return None
-
-    # Format the stack trace
     tb_str = "".join(traceback.format_exception(type(exc), exc, exc.__traceback__))
 
-    # Extract file/line from traceback
     tb_obj = exc.__traceback__
     affected_file = "unknown"
     affected_line = 0
@@ -139,9 +117,18 @@ def capture_error(
         affected_file = tb_obj.tb_frame.f_code.co_filename
         affected_line = tb_obj.tb_lineno
 
-    git_ctx = _get_git_context()
+    return tb_str, affected_file, affected_line
 
-    # 1. Log to event_log
+
+def _log_error_event(
+    exc: Exception,
+    source: str,
+    error_hash: str,
+    affected_file: str,
+    affected_line: int,
+    git_ctx: dict,
+) -> None:
+    """Log the captured error to the event log."""
     try:
         from swarm.event_log import EventType, log_event
 
@@ -161,8 +148,18 @@ def capture_error(
     except Exception as log_exc:
         logger.debug("Failed to log error event: %s", log_exc)
 
-    # 2. Create bug report task
-    task_id = None
+
+def _create_bug_report(
+    exc: Exception,
+    source: str,
+    context: dict | None,
+    error_hash: str,
+    tb_str: str,
+    affected_file: str,
+    affected_line: int,
+    git_ctx: dict,
+) -> str | None:
+    """Create a bug report task and return the task ID (or None on failure)."""
     try:
         from swarm.task_queue.models import create_task
 
@@ -195,7 +192,6 @@ def capture_error(
         )
         task_id = task.id
 
-        # Log the creation event
         try:
             from swarm.event_log import EventType, log_event
 
@@ -210,12 +206,16 @@ def capture_error(
             )
         except Exception as exc:
             logger.warning("Bug report screenshot error: %s", exc)
-            pass
+
+        return task_id
 
     except Exception as task_exc:
         logger.debug("Failed to create bug report task: %s", task_exc)
+        return None
 
-    # 3. Send notification
+
+def _notify_bug_report(exc: Exception, source: str) -> None:
+    """Send a push notification about the captured error."""
     try:
         from infrastructure.notifications.push import notifier
 
@@ -224,11 +224,12 @@ def capture_error(
             message=f"{type(exc).__name__} in {source}: {str(exc)[:80]}",
             category="system",
         )
-    except Exception as exc:
-        logger.warning("Bug report notification error: %s", exc)
-        pass
+    except Exception as notify_exc:
+        logger.warning("Bug report notification error: %s", notify_exc)
 
-    # 4. Record in session logger (via registered callback)
+
+def _record_to_session(exc: Exception, source: str) -> None:
+    """Record the error via the registered session callback."""
     if _error_recorder is not None:
         try:
             _error_recorder(
@@ -238,4 +239,50 @@ def capture_error(
         except Exception as log_exc:
             logger.warning("Bug report session logging error: %s", log_exc)
 
+
+def capture_error(
+    exc: Exception,
+    source: str = "unknown",
+    context: dict | None = None,
+) -> str | None:
+    """Capture an error and optionally create a bug report.
+
+    Args:
+        exc: The exception to capture
+        source: Module/component where the error occurred
+        context: Optional dict of extra context (request path, etc.)
+
+    Returns:
+        Task ID of the created bug report, or None if deduplicated/disabled
+    """
+    from config import settings
+
+    if not settings.error_feedback_enabled:
+        return None
+
+    error_hash = _stack_hash(exc)
+
+    if _is_duplicate(error_hash):
+        logger.debug("Duplicate error suppressed: %s (hash=%s)", exc, error_hash)
+        return None
+
+    tb_str, affected_file, affected_line = _extract_traceback_info(exc)
+    git_ctx = _get_git_context()
+
+    _log_error_event(exc, source, error_hash, affected_file, affected_line, git_ctx)
+
+    task_id = _create_bug_report(
+        exc,
+        source,
+        context,
+        error_hash,
+        tb_str,
+        affected_file,
+        affected_line,
+        git_ctx,
+    )
+
+    _notify_bug_report(exc, source)
+    _record_to_session(exc, source)
+
     return task_id

tests/infrastructure/test_error_capture.py

@@ -5,9 +5,14 @@ from datetime import UTC, datetime, timedelta
 from unittest.mock import patch
 
 from infrastructure.error_capture import (
+    _create_bug_report,
     _dedup_cache,
+    _extract_traceback_info,
     _get_git_context,
     _is_duplicate,
+    _log_error_event,
+    _notify_bug_report,
+    _record_to_session,
     _stack_hash,
     capture_error,
 )
@@ -193,3 +198,91 @@ class TestCaptureError:
 
     def teardown_method(self):
         _dedup_cache.clear()
+
+
+class TestExtractTracebackInfo:
+    """Test _extract_traceback_info helper."""
+
+    def test_returns_three_tuple(self):
+        try:
+            raise ValueError("extract test")
+        except ValueError as e:
+            tb_str, affected_file, affected_line = _extract_traceback_info(e)
+        assert "ValueError" in tb_str
+        assert "extract test" in tb_str
+        assert affected_file.endswith(".py")
+        assert affected_line > 0
+
+    def test_file_points_to_raise_site(self):
+        try:
+            _make_exception()
+        except ValueError as e:
+            _, affected_file, _ = _extract_traceback_info(e)
+        assert "test_error_capture" in affected_file
+
+
+class TestLogErrorEvent:
+    """Test _log_error_event helper."""
+
+    def test_does_not_crash_on_missing_deps(self):
+        try:
+            raise RuntimeError("log test")
+        except RuntimeError as e:
+            _log_error_event(e, "test", "abc123", "file.py", 42, {"branch": "main"})
+
+
+class TestCreateBugReport:
+    """Test _create_bug_report helper."""
+
+    def test_does_not_crash_on_missing_deps(self):
+        try:
+            raise RuntimeError("report test")
+        except RuntimeError as e:
+            result = _create_bug_report(
+                e, "test", None, "abc123", "traceback...", "file.py", 42, {}
+            )
+        # May return None if swarm deps unavailable — that's fine
+        assert result is None or isinstance(result, str)
+
+    def test_with_context(self):
+        try:
+            raise RuntimeError("ctx test")
+        except RuntimeError as e:
+            result = _create_bug_report(e, "test", {"path": "/api"}, "abc", "tb", "f.py", 1, {})
+        assert result is None or isinstance(result, str)
+
+
+class TestNotifyBugReport:
+    """Test _notify_bug_report helper."""
+
+    def test_does_not_crash(self):
+        try:
+            raise RuntimeError("notify test")
+        except RuntimeError as e:
+            _notify_bug_report(e, "test")
+
+
+class TestRecordToSession:
+    """Test _record_to_session helper."""
+
+    def test_does_not_crash_without_recorder(self):
+        try:
+            raise RuntimeError("session test")
+        except RuntimeError as e:
+            _record_to_session(e, "test")
+
+    def test_calls_registered_recorder(self):
+        from infrastructure.error_capture import register_error_recorder
+
+        calls = []
+        register_error_recorder(lambda **kwargs: calls.append(kwargs))
+        try:
+            try:
+                raise RuntimeError("callback test")
+            except RuntimeError as e:
+                _record_to_session(e, "test_source")
+            assert len(calls) == 1
+            assert "RuntimeError" in calls[0]["error"]
+            assert calls[0]["context"] == "test_source"
+        finally:
+            register_error_recorder(None)