-
Notifications
You must be signed in to change notification settings - Fork 7.8k
fix: add durable idempotency guard for tool retries (#5802) #6492
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
base: main
Are you sure you want to change the base?
Changes from all commits
d1d087f
d85df7a
303b79e
af4775b
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -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 | ||
|
coderabbitai[bot] marked this conversation as resolved.
|
||
|
|
||
| return wrapper | ||
|
|
||
| return decorator | ||
| Original file line number | Diff line number | Diff line change | ||||
|---|---|---|---|---|---|---|
| @@ -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 | ||||||
|
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. 🎯 Functional Correctness | 🔴 Critical | ⚡ Quick win Incorrect import path will fail pytest at repo root.
🐛 Proposed fix-from idempotency import IdempotencyGuard, idempotent
+from crewai.tools.idempotency import IdempotencyGuard, idempotent📝 Committable suggestion
Suggested change
🤖 Prompt for AI Agents |
||||||
|
|
||||||
|
|
||||||
| 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" | ||||||
|
Comment on lines
+75
to
+95
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. 🩺 Stability & Availability | 🟠 Major | ⚡ Quick win File backend persistence test doesn't actually test persistence.
To prove this, clearing The test needs to either:
This was previously flagged and marked as addressed, but the same instance-attribute pattern persists in the current code. 🧪 Proposed fix: clear class storage to force file-based lookup guard1.record(call_key, "payment_sent")
+ # Clear class-level storage to force file-based lookup
+ IdempotencyGuard._storage = {}
+
# 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)
+ # guard2 should load from file automatically (requires _load() in implementation)
+ # OR manually load into class storage:
+ import json
+ if storage_path.exists():
+ with open(storage_path, "r") as f:
+ IdempotencyGuard._storage = json.load(f)
assert guard2.is_duplicate(call_key)
assert guard2.get_cached_result(call_key) == "payment_sent"🧰 Tools🪛 ast-grep (0.44.1)[warning] 90-90: File path is request-/variable-derived; validate and normalize to prevent path traversal. (open-filename-from-request) 🤖 Prompt for AI Agents |
||||||
|
|
||||||
|
|
||||||
| 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 | ||||||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
📐 Maintainability & Code Quality | 🟠 Major | ⚡ Quick win
Missing
functools.wrapsbreaks downstream@tooldecorator.The decorator returns a bare
wrapperthat losesfunc.__name__,__doc__, and__module__. Since the usage example stacks@toolon top of@idempotent, the@tooldecorator will seewrapper.__name__ == "wrapper"instead of the original function name, likely causing incorrect tool registration.🔧 Proposed fix
📝 Committable suggestion
🤖 Prompt for AI Agents