From 7bb6f15c33d7f282f7fade3f0a2807bb62516d27 Mon Sep 17 00:00:00 2001 From: kimi Date: Thu, 19 Mar 2026 20:04:33 -0400 Subject: [PATCH] refactor: break up capture_error() into testable helpers Extract 5 focused helpers from the 138-line capture_error(): - _extract_origin(): walk traceback for file/line - _log_error_event(): log to event log (best-effort) - _create_bug_report(): create task and log creation event - _send_error_notification(): push notification - _record_to_session(): forward to session recorder capture_error() now orchestrates the helpers in ~25 lines. Added tests for each new helper. Fixes #506 Co-Authored-By: Claude Opus 4.6 --- src/infrastructure/error_capture.py | 127 +++++++++++++-------- tests/infrastructure/test_error_capture.py | 89 +++++++++++++++ 2 files changed, 167 insertions(+), 49 deletions(-) diff --git a/src/infrastructure/error_capture.py b/src/infrastructure/error_capture.py index ee75afb..835454e 100644 --- a/src/infrastructure/error_capture.py +++ b/src/infrastructure/error_capture.py @@ -100,48 +100,25 @@ 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.) - - 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 - - # Format the stack trace - tb_str = "".join(traceback.format_exception(type(exc), exc, exc.__traceback__)) - - # Extract file/line from traceback +def _extract_origin(exc: Exception) -> tuple[str, int]: + """Walk the traceback to find the deepest file and line number.""" tb_obj = exc.__traceback__ - affected_file = "unknown" - affected_line = 0 while tb_obj and tb_obj.tb_next: tb_obj = tb_obj.tb_next if tb_obj: - affected_file = tb_obj.tb_frame.f_code.co_filename - affected_line = tb_obj.tb_lineno + return tb_obj.tb_frame.f_code.co_filename, tb_obj.tb_lineno + return "unknown", 0 - git_ctx = _get_git_context() - # 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 error to the event log (best-effort).""" try: from swarm.event_log import EventType, log_event @@ -161,8 +138,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, + error_hash: str, + affected_file: str, + affected_line: int, + git_ctx: dict, + tb_str: str, + context: dict | None, +) -> str | None: + """Create a bug report task and return its ID (best-effort).""" try: from swarm.task_queue.models import create_task @@ -193,29 +180,30 @@ def capture_error( auto_approve=True, task_type="bug_report", ) - task_id = task.id - # Log the creation event try: from swarm.event_log import EventType, log_event log_event( EventType.BUG_REPORT_CREATED, source=source, - task_id=task_id, + task_id=task.id, data={ "error_hash": error_hash, "title": title[:100], }, ) - except Exception as exc: - logger.warning("Bug report screenshot error: %s", exc) - pass + except Exception as log_exc: + logger.warning("Bug report log error: %s", log_exc) + 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 _send_error_notification(exc: Exception, source: str) -> None: + """Push a notification about the captured error (best-effort).""" try: from infrastructure.notifications.push import notifier @@ -224,11 +212,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: + """Forward the error to the registered session recorder (best-effort).""" if _error_recorder is not None: try: _error_recorder( @@ -238,4 +227,44 @@ 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 = "".join(traceback.format_exception(type(exc), exc, exc.__traceback__)) + affected_file, affected_line = _extract_origin(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, error_hash, affected_file, affected_line, git_ctx, tb_str, context + ) + + _send_error_notification(exc, source) + _record_to_session(exc, source) + return task_id diff --git a/tests/infrastructure/test_error_capture.py b/tests/infrastructure/test_error_capture.py index ddf369c..a396f24 100644 --- a/tests/infrastructure/test_error_capture.py +++ b/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_origin, _get_git_context, _is_duplicate, + _log_error_event, + _record_to_session, + _send_error_notification, _stack_hash, capture_error, ) @@ -193,3 +198,87 @@ class TestCaptureError: def teardown_method(self): _dedup_cache.clear() + + +class TestExtractOrigin: + """Test _extract_origin helper.""" + + def test_returns_file_and_line(self): + try: + _make_exception() + except ValueError as e: + filename, lineno = _extract_origin(e) + assert filename.endswith("test_error_capture.py") + assert lineno > 0 + + def test_no_traceback_returns_defaults(self): + exc = ValueError("no tb") + exc.__traceback__ = None + assert _extract_origin(exc) == ("unknown", 0) + + +class TestLogErrorEvent: + """Test _log_error_event helper.""" + + def test_does_not_crash_when_event_log_missing(self): + try: + raise RuntimeError("log test") + except RuntimeError as e: + _log_error_event(e, "test", "abc123", "file.py", 42, {}) + + +class TestCreateBugReport: + """Test _create_bug_report helper.""" + + def test_returns_none_on_import_failure(self): + try: + raise RuntimeError("report test") + except RuntimeError as e: + with patch("infrastructure.error_capture.logger"): + result = _create_bug_report(e, "test", "abc", "f.py", 1, {}, "tb", None) + # Returns a task id or None depending on whether swarm is available + assert result is None or isinstance(result, str) + + +class TestSendErrorNotification: + """Test _send_error_notification helper.""" + + def test_does_not_crash_on_notifier_failure(self): + try: + raise RuntimeError("notify test") + except RuntimeError as e: + _send_error_notification(e, "test") + + +class TestRecordToSession: + """Test _record_to_session helper.""" + + def test_noop_when_no_recorder(self): + import infrastructure.error_capture as ec + + original = ec._error_recorder + try: + ec._error_recorder = None + try: + raise RuntimeError("session test") + except RuntimeError as e: + _record_to_session(e, "test") # should not crash + finally: + ec._error_recorder = original + + def test_calls_registered_recorder(self): + import infrastructure.error_capture as ec + + original = ec._error_recorder + calls = [] + try: + ec._error_recorder = lambda **kwargs: calls.append(kwargs) + try: + raise RuntimeError("recorded") + except RuntimeError as e: + _record_to_session(e, "src") + assert len(calls) == 1 + assert "RuntimeError: recorded" in calls[0]["error"] + assert calls[0]["context"] == "src" + finally: + ec._error_recorder = original