feat: SwarmValidationError for handoff schema validation

langgraph_swarm/__init__.py

@@ -1,8 +1,11 @@
+from langgraph_swarm.errors import SwarmValidationError, TaskExecutionError
 from langgraph_swarm.handoff import create_handoff_tool
 from langgraph_swarm.swarm import SwarmState, add_active_agent_router, create_swarm
 
 __all__ = [
     "SwarmState",
+    "SwarmValidationError",
+    "TaskExecutionError",
     "add_active_agent_router",
     "create_handoff_tool",
     "create_swarm",

langgraph_swarm/errors.py

@@ -0,0 +1,32 @@
+"""Custom exceptions for LangGraph Swarm."""
+
+
+class SwarmValidationError(Exception):
+    """Raised when human feedback validation fails in HITL gates.
+
+    This exception is used to signal validation failures when processing
+    handoff payloads in human-in-the-loop workflows. It helps distinguish
+    schema/validation errors from other runtime exceptions.
+
+    Example:
+        >>> if "path_validations" not in feedback:
+        ...     raise SwarmValidationError("Missing path_validations key")
+    """
+
+    pass
+
+
+class TaskExecutionError(Exception):
+    """Raised when a worker task execution fails.
+
+    This exception wraps worker-level failures to provide cleaner
+    error handling and recovery in swarm orchestration.
+
+    Example:
+        >>> try:
+        ...     result = worker.execute(task)
+        ... except Exception as e:
+        ...     raise TaskExecutionError(f"Worker failed: {e}") from e
+    """
+
+    pass

tests/test_validation_error.py

@@ -0,0 +1,62 @@
+"""Tests for SwarmValidationError and TaskExecutionError exceptions."""
+
+import pytest
+
+from langgraph_swarm.errors import SwarmValidationError, TaskExecutionError
+
+
+class TestSwarmValidationError:
+    """Tests for SwarmValidationError exception."""
+
+    def test_swarm_validation_error_raises(self):
+        """Test that SwarmValidationError can be raised and caught."""
+        with pytest.raises(SwarmValidationError):
+            raise SwarmValidationError("Invalid feedback structure")
+
+    def test_swarm_validation_error_message(self):
+        """Test that error message is preserved."""
+        error = SwarmValidationError("Missing path_validations key")
+        assert "path_validations" in str(error)
+
+    def test_swarm_validation_error_inheritance(self):
+        """Test that SwarmValidationError inherits from Exception."""
+        error = SwarmValidationError("test")
+        assert isinstance(error, Exception)
+
+    def test_swarm_validation_error_with_cause(self):
+        """Test exception chaining works correctly."""
+        original = ValueError("bad value")
+        try:
+            raise SwarmValidationError("Validation failed") from original
+        except SwarmValidationError as e:
+            assert e.__cause__ is original
+
+
+class TestTaskExecutionError:
+    """Tests for TaskExecutionError exception."""
+
+    def test_task_execution_error_raises(self):
+        """Test that TaskExecutionError can be raised and caught."""
+        with pytest.raises(TaskExecutionError):
+            raise TaskExecutionError("Worker failed")
+
+    def test_task_execution_error_message(self):
+        """Test that error message is preserved."""
+        error = TaskExecutionError("Task timeout after 30s")
+        assert "timeout" in str(error)
+
+    def test_task_execution_error_inheritance(self):
+        """Test that TaskExecutionError inherits from Exception."""
+        error = TaskExecutionError("test")
+        assert isinstance(error, Exception)
+
+
+class TestErrorImports:
+    """Test that errors are importable from main package."""
+
+    def test_import_from_package(self):
+        """Test errors can be imported from langgraph_swarm."""
+        from langgraph_swarm import SwarmValidationError, TaskExecutionError
+
+        assert SwarmValidationError is not None
+        assert TaskExecutionError is not None