diff --git a/lib/crewai/src/crewai/tools/idempotency/__init__.py b/lib/crewai/src/crewai/tools/idempotency/__init__.py
new file mode 100644
index 0000000000..89bccd2a24
--- /dev/null
+++ b/lib/crewai/src/crewai/tools/idempotency/__init__.py
@@ -0,0 +1,143 @@
+"""
+Fix for CrewAI #5802: Tool re-execution on task retry has no idempotency guard
+
+Root cause:
+  _check_tool_repeated_usage() only checks the last used tool in memory.
+  When a task fails and retries, tools_handler is reset, so last_used_tool becomes None.
+  The duplicate tool call executes again — causing duplicate payments/emails/trades.
+
+Fix:
+  Add a durable idempotency layer that persists across task retries.
+  Each tool call gets a stable hash (tool_name + arguments).
+  If the hash exists in durable storage, return cached result instead of re-executing.
+
+Usage:
+  from crewai.tools.idempotency import IdempotencyGuard, idempotent
+
+  @tool
+  @idempotent(storage_backend="file")
+  def send_payment(amount: float, recipient: str) -> str:
+      result = stripe.charge(amount, recipient)
+      return result
+"""
+
+from __future__ import annotations
+import hashlib
+import json
+from typing import Any, Dict, Optional, Tuple
+from pathlib import Path
+
+
+class IdempotencyGuard:
+    """
+    Durable idempotency guard for CrewAI tools.
+
+    Survives task retries, worker re-dispatch, and process restarts.
+    Based on CCS (Conformance Checking Standard) Identity dimension.
+    """
+
+    # Class-level cache (shared within a process)
+    _storage: Dict[str, Any] = {}
+    # File path for durable backend (shared across all instances using file backend)
+    _storage_path: Optional[Path] = None
+
+    def __init__(self, tool_name: str, storage_backend: str = "memory"):
+        """
+        Args:
+            tool_name: Name of the tool (used in hash computation)
+            storage_backend: "memory" (default) or "file" (durable)
+        """
+        self.tool_name = tool_name
+        self.storage_backend = storage_backend
+
+        if storage_backend == "file":
+            IdempotencyGuard._storage_path = Path(".ccs_idempotency.json")
+            self._load()
+
+    def _load(self):
+        """Load storage from file (if using file backend)."""
+        if (
+            self.storage_backend == "file"
+            and IdempotencyGuard._storage_path
+            and IdempotencyGuard._storage_path.exists()
+        ):
+            with open(IdempotencyGuard._storage_path, "r") as f:
+                IdempotencyGuard._storage = json.load(f)
+
+    def _compute_hash(self, call_key: Dict[str, Any]) -> str:
+        """Compute stable hash from tool_name + call key (args + kwargs)"""
+        key = json.dumps({
+            "tool": self.tool_name,
+            "call_key": call_key
+        }, sort_keys=True, default=str)
+        return hashlib.sha256(key.encode()).hexdigest()[:16]
+
+    def _persist(self):
+        """Persist storage to disk (if using file backend)"""
+        if self.storage_backend == "file" and IdempotencyGuard._storage_path:
+            with open(IdempotencyGuard._storage_path, "w") as f:
+                json.dump(IdempotencyGuard._storage, f)
+
+    def is_duplicate(self, call_key: Dict[str, Any]) -> bool:
+        """Check if this tool call has already been executed"""
+        h = self._compute_hash(call_key)
+        return h in IdempotencyGuard._storage
+
+    def get_cached_result(self, call_key: Dict[str, Any]) -> Any:
+        """Get cached result for duplicate call"""
+        h = self._compute_hash(call_key)
+        return IdempotencyGuard._storage.get(h)
+
+    def record(self, call_key: Dict[str, Any], result: Any):
+        """Record tool execution result"""
+        h = self._compute_hash(call_key)
+        IdempotencyGuard._storage[h] = result
+        self._persist()
+
+    @classmethod
+    def reset(cls):
+        """Reset storage (for testing)"""
+        cls._storage = {}
+        if cls._storage_path and cls._storage_path.exists():
+            cls._storage_path.unlink()
+
+
+def idempotent(storage_backend: str = "memory"):
+    """
+    Decorator to make a tool idempotent.
+
+    Usage:
+        @tool
+        @idempotent(storage_backend="file")
+        def send_payment(amount: float, recipient: str) -> str:
+            stripe.charge(amount, recipient)
+            return "payment sent"
+    """
+    def decorator(func):
+        def wrapper(*args, **kwargs):
+            guard = IdempotencyGuard(
+                tool_name=func.__name__, storage_backend=storage_backend
+            )
+
+            # Build call key from BOTH positional and keyword arguments
+            call_key = {
+                "args": args,
+                "kwargs": {k: v for k, v in sorted(kwargs.items())},
+            }
+
+            if guard.is_duplicate(call_key):
+                cached = guard.get_cached_result(call_key)
+                print(
+                    f"[CCS] Blocked duplicate: {func.__name__}"
+                    f", returning cached result"
+                )
+                return cached
+
+            # Execute
+            result = func(*args, **kwargs)
+            guard.record(call_key, result)
+            return result
+
+        return wrapper
+
+    return decorator
diff --git a/lib/crewai/src/crewai/tools/idempotency/test_idempotency.py b/lib/crewai/src/crewai/tools/idempotency/test_idempotency.py
new file mode 100644
index 0000000000..ad94017d08
--- /dev/null
+++ b/lib/crewai/src/crewai/tools/idempotency/test_idempotency.py
@@ -0,0 +1,173 @@
+"""
+Tests for CrewAI #5802 fix: Idempotency guard for tool retries
+"""
+import pytest
+from pathlib import Path
+from idempotency import IdempotencyGuard, idempotent
+
+
+class TestIdempotencyGuard:
+    """Test IdempotencyGuard core functionality"""
+
+    def setup_method(self):
+        IdempotencyGuard.reset()
+
+    def test_first_execution_not_duplicate(self):
+        """First tool call should not be detected as duplicate"""
+        guard = IdempotencyGuard("send_payment")
+        call_key = {"args": (100, "alice"), "kwargs": {}}
+
+        assert not guard.is_duplicate(call_key)
+
+    def test_second_execution_is_duplicate(self):
+        """Second tool call with same args should be duplicate"""
+        guard = IdempotencyGuard("send_payment")
+        call_key = {"args": (), "kwargs": {"amount": 100, "recipient": "alice"}}
+
+        guard.record(call_key, "payment_sent")
+
+        assert guard.is_duplicate(call_key)
+        assert guard.get_cached_result(call_key) == "payment_sent"
+
+    def test_different_args_not_duplicate(self):
+        """Different arguments should not be duplicate"""
+        guard = IdempotencyGuard("send_payment")
+
+        guard.record({"args": (), "kwargs": {"amount": 100, "recipient": "alice"}}, "payment_1")
+
+        assert not guard.is_duplicate({"args": (), "kwargs": {"amount": 200, "recipient": "bob"}})
+
+    def test_positional_args_distinguished(self):
+        """Positional args must produce different keys (fix for CodeRabbit Critical)"""
+        guard = IdempotencyGuard("send_payment")
+
+        key1 = {"args": (100, "alice"), "kwargs": {}}
+        key2 = {"args": (200, "bob"), "kwargs": {}}
+
+        guard.record(key1, "payment_1")
+
+        assert guard.is_duplicate(key1)
+        assert not guard.is_duplicate(key2)
+
+    def test_mixed_args_and_kwargs(self):
+        """Mix of positional and keyword args should be handled correctly"""
+        guard = IdempotencyGuard("send_payment")
+
+        key1 = {"args": (100,), "kwargs": {"recipient": "alice"}}
+        key2 = {"args": (200,), "kwargs": {"recipient": "bob"}}
+        key3 = {"args": (100,), "kwargs": {"recipient": "alice"}}
+
+        guard.record(key1, "payment_1")
+
+        assert guard.is_duplicate(key3)  # Same args -> duplicate
+        assert not guard.is_duplicate(key2)  # Different args -> not duplicate
+
+    def test_different_tools_not_duplicate(self):
+        """Different tools with same args should not be duplicate"""
+        guard1 = IdempotencyGuard("send_payment")
+        guard2 = IdempotencyGuard("send_email")
+
+        call_key = {"args": (123,), "kwargs": {}}
+        guard1.record(call_key, "payment_sent")
+
+        assert not guard2.is_duplicate(call_key)
+
+    def test_file_backend_persists(self, tmp_path):
+        """File backend should persist across instances"""
+        storage_path = tmp_path / ".ccs_idempotency.json"
+
+        # First instance records
+        guard1 = IdempotencyGuard("send_payment", storage_backend="file")
+        guard1._storage_path = storage_path
+        call_key = {"args": (100, "alice"), "kwargs": {}}
+        guard1.record(call_key, "payment_sent")
+
+        # Second instance should see the record
+        guard2 = IdempotencyGuard("send_payment", storage_backend="file")
+        guard2._storage_path = storage_path
+        guard2._storage = {}  # simulate fresh load
+        import json
+        if storage_path.exists():
+            with open(storage_path, "r") as f:
+                guard2._storage = json.load(f)
+
+        assert guard2.is_duplicate(call_key)
+        assert guard2.get_cached_result(call_key) == "payment_sent"
+
+
+class TestIdempotentDecorator:
+    """Test @idempotent decorator"""
+
+    def setup_method(self):
+        IdempotencyGuard.reset()
+
+    def test_decorator_blocks_duplicate(self):
+        """Decorator should block duplicate calls with same args"""
+        call_count = 0
+
+        @idempotent()
+        def send_payment(amount, recipient):
+            nonlocal call_count
+            call_count += 1
+            return f"sent {amount} to {recipient}"
+
+        # First call executes
+        r1 = send_payment(100, "alice")
+        assert r1 == "sent 100 to alice"
+        assert call_count == 1
+
+        # Second call with same args returns cached
+        r2 = send_payment(100, "alice")
+        assert r2 == "sent 100 to alice"
+        assert call_count == 1  # Not executed again
+
+    def test_decorator_allows_different_args(self):
+        """Decorator should allow calls with different args"""
+        call_count = 0
+
+        @idempotent()
+        def send_payment(amount, recipient):
+            nonlocal call_count
+            call_count += 1
+            return f"sent {amount} to {recipient}"
+
+        r1 = send_payment(100, "alice")
+        r2 = send_payment(200, "bob")
+
+        assert r1 == "sent 100 to alice"
+        assert r2 == "sent 200 to bob"
+        assert call_count == 2  # Both executed
+
+    def test_decorator_with_kwargs_only(self):
+        """Decorator should work with keyword-only calls"""
+        call_count = 0
+
+        @idempotent()
+        def send_payment(amount=0, recipient=""):
+            nonlocal call_count
+            call_count += 1
+            return f"sent {amount} to {recipient}"
+
+        r1 = send_payment(amount=100, recipient="alice")
+        r2 = send_payment(amount=100, recipient="alice")
+
+        assert call_count == 1  # Second call blocked
+        assert r2 == r1
+
+    def test_decorator_positional_vs_kwargs(self):
+        """Decorator should distinguish positional from keyword calls"""
+        call_count = 0
+
+        @idempotent()
+        def send_payment(amount, recipient):
+            nonlocal call_count
+            call_count += 1
+            return f"sent {amount} to {recipient}"
+
+        # Positional
+        send_payment(100, "alice")
+        assert call_count == 1
+
+        # Same values as kwargs — should still be different call key
+        send_payment(amount=100, recipient="alice")
+        assert call_count == 2  # Different call key format