From d9df58e220aadcb26dee29a62ce52fd24bcabb41 Mon Sep 17 00:00:00 2001
From: Igor Holt <iholt@mymail.aacc.edu>
Date: Mon, 20 Apr 2026 01:03:53 -0400
Subject: [PATCH] feat: add SwarmValidationError for HITL handoff validation

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
Signed-off-by: Igor Holt <iholt@mymail.aacc.edu>
---
 langgraph_swarm/__init__.py    |  3 ++
 langgraph_swarm/errors.py      | 32 ++++++++++++++++++
 tests/test_validation_error.py | 62 ++++++++++++++++++++++++++++++++++
 3 files changed, 97 insertions(+)
 create mode 100644 langgraph_swarm/errors.py
 create mode 100644 tests/test_validation_error.py

diff --git a/langgraph_swarm/__init__.py b/langgraph_swarm/__init__.py
index b3f90c9..1f10483 100644
--- a/langgraph_swarm/__init__.py
+++ b/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",
diff --git a/langgraph_swarm/errors.py b/langgraph_swarm/errors.py
new file mode 100644
index 0000000..a28e2bb
--- /dev/null
+++ b/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
diff --git a/tests/test_validation_error.py b/tests/test_validation_error.py
new file mode 100644
index 0000000..e0e012c
--- /dev/null
+++ b/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