Unkillable Agent Engineering #2: The Idempotency Shield — Making Your Agent Execute Once, No Matter How Many Times It Retries

Prerequisites:

• [x] Observability deployed (07-19)
• [x] This is "Unkillable Agent Engineering" Series #2

---

At 3 AM, your Agent receives one instruction: "Send annual reports to all paid users."

It starts processing—querying the database, generating reports, sending emails. 998 emails out of 1,200 are sent.

Then your server forces a restart.

The Agent wakes up, looks at its to-do list: "202 emails remaining." It continues.

Result: 1,400 emails sent to the same user base—998 of them are duplicates.

Your clients call to ask: "What kind of garbage is this?"

This isn't a bug. This is missing idempotency.

---

1. Why Agents Are More Vulnerable to Idempotency Disasters Than Regular Programs

Traditional programs: You press one button, it runs once.
Agent: You press one button, it calls 17 tools, accesses 3 APIs, and might restart 5 times mid-process.

The root cause: Agent execution granularity is far larger than "one function call."

Scenario	Non-Idempotent Consequence	Typical Loss
Payment request retry	Double charge	$2,000~50,000
Email broadcast restart	Duplicate sends	User complaints + unsubscribes
Database write retry	Duplicate records	Data pollution
API call timeout retry	Duplicate resource creation	Resource waste
Task interruption recovery	Partial task runs twice	Logic errors

---

2. Three-Layer Idempotency Defense Architecture

┌─────────────────────────────────────────────────────────┐
│                    IdempotencyGuard                      │
├───────────────┬─────────────────┬─────────────────────┤
│  Layer 1      │   Layer 2        │   Layer 3           │
│  Request      │   Execution      │   Result            │
│  Deduplication │  Idempotency     │   Idempotency       │
│               │                  │                     │
│  IdempotencyKey│ RequestDedupe   │ ResultIdempotent    │
│               │                  │                     │
│  Entry Guard  │  Process Safety  │ Output Guarantee     │
└───────────────┴─────────────────┴─────────────────────┘

---

3. Layer 1: IdempotencyKey Request Deduplication

import hashlib
import time
from datetime import datetime, timedelta
from typing import Optional, Any, Callable
from dataclasses import dataclass, field
import json

@dataclass
class IdempotencyRecord:
    """Idempotency record"""
    key: str
    status: str  # processing | completed | failed
    result: Optional[Any] = None
    created_at: datetime = field(default_factory=datetime.now)
    expires_at: datetime = None

    def __post_init__(self):
        if self.expires_at is None:
            self.expires_at = self.created_at + timedelta(hours=24)

class IdempotencyGuard:
    """
    Idempotency Guardian — Ensures Agent operations execute only once,
    regardless of how many times they are retried.

    Use cases:
    - Payment/refund requests
    - Resource creation operations
    - State changes
    - External API calls
    """

    def __init__(self, storage=None, ttl_seconds: int = 86400):
        """
        Args:
            storage: Storage backend (Redis/Dict/custom supported)
            ttl_seconds: Idempotency key TTL, default 24 hours
        """
        self.storage = storage or {}  # In-memory for dev; use Redis in prod
        self.ttl = ttl_seconds
        self._lock_timeout = 30  # seconds

    def generate_key(self, agent_id: str, operation: str, params: dict) -> str:
        """
        Generate idempotency key

        Formula: hash(agent_id + operation + sorted_params)
        """
        normalized = json.dumps(params, sort_keys=True, default=str)
        raw = f"{agent_id}:{operation}:{normalized}"
        return hashlib.sha256(raw.encode()).hexdigest()[:32]

    def check_and_lock(self, key: str) -> tuple[bool, Optional[Any]]:
        """
        Check idempotency status

        Returns:
            (is_new_request, existing_result)
            - is_new_request=True: New request, safe to execute
            - is_new_request=False, existing_result has value: Already ran, return cached result
            - is_new_request=False, existing_result=None: Currently processing
        """
        now = datetime.now()

        if key in self.storage:
            record = self.storage[key]

            if now > record.expires_at:
                del self.storage[key]
                return True, None

            if record.status == "completed":
                return False, record.result
            elif record.status == "processing":
                return False, None

        return True, None

    def mark_processing(self, key: str) -> bool:
        """Mark as processing"""
        self.storage[key] = IdempotencyRecord(
            key=key,
            status="processing",
            expires_at=datetime.now() + timedelta(seconds=self.ttl)
        )
        return True

    def mark_completed(self, key: str, result: Any):
        """Mark as completed"""
        if key in self.storage:
            self.storage[key].status = "completed"
            self.storage[key].result = result

    def mark_failed(self, key: str):
        """Mark as failed"""
        if key in self.storage:
            self.storage[key].status = "failed"

    def execute(self, 
                 agent_id: str,
                 operation: str, 
                 params: dict, 
                 func: Callable[[], Any],
                 allow_retry: bool = True) -> Any:
        """
        Execute with idempotency guarantee

        Args:
            agent_id: Agent identifier
            operation: Operation name
            params: Operation parameters
            func: Actual execution function
            allow_retry: Allow retry on failure

        Returns:
            func result (possibly cached)
        """
        key = self.generate_key(agent_id, operation, params)

        is_new, cached_result = self.check_and_lock(key)

        if not is_new:
            if cached_result is not None:
                print(f"[IdempotencyGuard] Returning cached result for key={key}")
                return cached_result
            else:
                print(f"[IdempotencyGuard] Request in progress for key={key}")
                raise Exception("Operation already in progress")

        self.mark_processing(key)

        try:
            result = func()
            self.mark_completed(key, result)
            print(f"[IdempotencyGuard] Execution completed for key={key}")
            return result
        except Exception as e:
            self.mark_failed(key)
            print(f"[IdempotencyGuard] Execution failed for key={key}: {e}")
            if not allow_retry:
                raise
            raise

---

4. Layer 2: Request Deduplication Middleware

from functools import wraps
import asyncio
from typing import TypeVar, ParamSpec

P = ParamSpec('P')
T = TypeVar('T')

class IdempotencyMiddleware:
    """Idempotency Middleware — Decorator form"""

    def __init__(self, guard: IdempotencyGuard):
        self.guard = guard

    def idempotent(self, 
                   agent_id: str,
                   operation: str,
                   ttl_seconds: int = 3600):
        """
        Idempotency decorator

        Usage:
        ```

python
        guard = IdempotencyGuard()
        middleware = IdempotencyMiddleware(guard)

        @middleware.idempotent(
            agent_id="my-agent",
            operation="send_email"
        )
        async def send_email_task(recipients: list, subject: str):
            # Actual sending logic
            return await email_service.send(recipients, subject)


        ```
        """
        def decorator(func: Callable[P, T]) -> Callable[P, T]:
            @wraps(func)
            async def async_wrapper(*args: P.args, **kwargs: P.kwargs) -> T:
                idempotency_params = {
                    "args": str(args),
                    "kwargs": str(sorted(kwargs.items()))
                }

                key = self.guard.generate_key(
                    agent_id, 
                    operation, 
                    idempotency_params
                )

                is_new, cached = self.guard.check_and_lock(key)

                if not is_new and cached is not None:
                    return cached

                self.guard.mark_processing(key)

                try:
                    result = await func(*args, **kwargs)
                    self.guard.mark_completed(key, result)
                    return result
                except Exception as e:
                    self.guard.mark_failed(key)
                    raise

            @wraps(func)
            def sync_wrapper(*args: P.args, **kwargs: P.kwargs) -> T:
                idempotency_params = {
                    "args": str(args),
                    "kwargs": str(sorted(kwargs.items()))
                }

                key = self.guard.generate_key(
                    agent_id, 
                    operation, 
                    idempotency_params
                )

                is_new, cached = self.guard.check_and_lock(key)

                if not is_new and cached is not None:
                    return cached

                self.guard.mark_processing(key)

                try:
                    result = func(*args, **kwargs)
                    self.guard.mark_completed(key, result)
                    return result
                except Exception as e:
                    self.guard.mark_failed(key)
                    raise

            return async_wrapper if asyncio.iscoroutinefunction(func) else sync_wrapper

        return decorator

---

5. Layer 3: Natural Idempotency — Designing Self-Idempotent Operations

Not all operations need idempotency guards. Designing naturally idempotent operations is a better strategy:

class NaturallyIdempotent:
    """
    Natural Idempotency Library — These operations don't produce side effects
    when run multiple times.

    Design Principles:
    1. UPSERT > INSERT/UPDATE (update if exists, create if not)
    2. DELETE is idempotent: deleting non-existent things doesn't error
    3. GET operations: naturally idempotent
    4. SET operations: overwrite old value, always maintain last state
    """

    # ✅ Naturally idempotent operations
    NATURALLY_IDEMPOTENT = [
        "GET",      # Read — always idempotent
        "SET",      # Set — overwrites old value
        "UPSERT",   # Update or insert
        "DELETE",   # Deleting non-existent doesn't error
        "COUNT",    # Count — naturally idempotent
        "EXISTS",   # Existence check — naturally idempotent
    ]

    # ⚠️ Operations needing guard
    NEEDS_GUARD = [
        "CREATE",   # Creation — errors on duplicate
        "SEND",     # Sending — duplicates on retry
        "PAY",      # Payment — charges double on retry
        "REFUND",   # Refund — refunds double on retry
        "SUBSCRIBE",# Subscription — double subscribes on retry
        "CHARGE",   # Charge — charges double on retry
    ]

    @classmethod
    def design_upsert(cls, 
                      table: str, 
                      unique_key: dict, 
                      data: dict) -> str:
        """
        Generate idempotent UPSERT SQL

        ```

sql
        -- Example: User subscription status update
        UPDATE subscriptions 
        SET status = 'active', updated_at = NOW()
        WHERE user_id = '123'

        INSERT INTO subscriptions (user_id, status, created_at, updated_at)
        SELECT '123', 'active', NOW(), NOW()
        WHERE NOT EXISTS (
            SELECT 1 FROM subscriptions WHERE user_id = '123'
        )


        ```
        """
        pass  # Adjust based on your database

    @classmethod
    def needs_guard(cls, operation: str) -> bool:
        """Check if operation needs idempotency guard"""
        return operation.upper() in cls.NEEDS_GUARD

---

6. Complete Agent Idempotency Integration Example

# agent_idempotency_example.py
"""
Complete Agent Idempotency Integration Example
Scenario: User requests Agent to notify all customers
"""

guard = IdempotencyGuard()
middleware = IdempotencyMiddleware(guard)

class NotificationAgent:
    def __init__(self):
        self.guard = IdempotencyGuard()
        self.middleware = IdempotencyMiddleware(self.guard)
        self.sent_records = {}  # Mock database

    def _generate_batch_id(self, batch_name: str, recipients: list) -> str:
        """Generate batch ID"""
        return hashlib.sha256(
            f"{batch_name}:{len(recipients)}".encode()
        ).hexdigest()[:16]

    @middleware.idempotent(
        agent_id="notification-agent-v1",
        operation="send_batch_notification"
    )
    def send_batch_notification(self, batch_name: str, recipients: list, message: str):
        """
        Batch notification sending — Idempotency guaranteed

        No matter how many times this is called, each user receives only one notification
        """
        batch_id = self._generate_batch_id(batch_name, recipients)

        print(f"[Agent] Starting batch {batch_id}, {len(recipients)} recipients")

        sent = 0
        for recipient in recipients:
            # Check if already sent (Layer 2 protection)
            if recipient in self.sent_records:
                print(f"[Agent] {recipient} already sent, skipping")
                continue

            self._send_single(recipient, message)
            self.sent_records[recipient] = batch_id
            sent += 1

        result = {
            "batch_id": batch_id,
            "sent_count": sent,
            "total": len(recipients),
            "skipped": len(recipients) - sent
        }

        print(f"[Agent] Batch {batch_id} complete: {sent}/{len(recipients)}")
        return result

    def _send_single(self, recipient: str, message: str):
        """Send single notification"""
        print(f"[Agent] Sending to {recipient}: {message[:20]}...")


# Usage
agent = NotificationAgent()

# First call — normal send
result1 = agent.send_batch_notification(
    batch_name="monthly_report",
    recipients=["user1", "user2", "user3"],
    message="Your monthly report is ready..."
)
print(f"First call: {result1}")

# Simulate restart + retry — returns cached result, no duplicate send
result2 = agent.send_batch_notification(
    batch_name="monthly_report",
    recipients=["user1", "user2", "user3"],
    message="Your monthly report is ready..."
)
print(f"Second call (retry): {result2}")

# Verification: same results, no duplicate sends
assert result1 == result2

---

7. Idempotency Deployment Checklist

Pre-deployment checks:

• [ ] All payment/refund operations use idempotency guard
• [ ] All creation operations use UPSERT instead of INSERT
• [ ] All external API calls have idempotency keys
• [ ] Task restart recovery logic is idempotent
• [ ] Database transaction rollbacks don't affect idempotency state
• [ ] Idempotency keys have reasonable TTL (24h is usually sufficient)
• [ ] Idempotency guard itself is thread-safe

---

8. Integration with Unkillable Agent Engineering

The Idempotency Guard is one core component of "Unkillable Agent Engineering":

Layer	Component	Responsibility
Observability	TraceGuard	Discover problems
Idempotency	IdempotencyGuard	Ensure no duplicate operations
Assertion	AssertionGuard	Verify result correctness
Drift Detection	DriftDetector	Detect behavior drift
Circuit Breaker	CircuitBreaker	Prevent cascade failures

These five layers work together to form a complete Agent survival system.

This is also the core thinking behind how we built the idempotency module in ARK Trust — not remediation after the fact, but ensuring every operation is safely re-executable by design.

---

Further Reading:

• "Unkillable Agent Engineering #1: The Three Pillars of Observability"
• "Seven Ways Your AI Agent Dies"