AI agent idempotency design: safe retries and duplicate tool call prevention

• Transient execution error: The tool call partially executed (e.g., email sent but delivery confirmation not received) and threw an exception. The model calls the tool again with the same arguments, causing a second send.
• Confusing error message: The tool succeeded but returned an error-like message (e.g., “0 rows updated” when the record already had the target value). The model interprets this as failure and retries.
• Tool-call loop: The tool returned a result the model did not know how to handle, causing it to cycle through a set of tool calls repeatedly, re-executing each one every cycle.

import sqlite3
from dataclasses import dataclass

@dataclass
class ToolResult:
    success: bool
    message: str
    created: bool  # True if new, False if already existed

def upsert_user_record(
    db_path: str,
    user_id: str,
    email: str,
    plan: str = "trial",
) -> ToolResult:
    """
    Idempotent user creation: safe to call multiple times with the same user_id.
    Returns success=True whether the row was created or already existed.
    The model never sees a 'duplicate' error, so it does not retry.
    """
    with sqlite3.connect(db_path) as conn:
        existing = conn.execute(
            "SELECT user_id FROM users WHERE user_id = ?", (user_id,)
        ).fetchone()

        if existing:
            # Already exists: update email/plan if changed, return created=False
            conn.execute(
                "UPDATE users SET email=?, plan=? WHERE user_id=?",
                (email, plan, user_id),
            )
            return ToolResult(
                success=True,
                message=f"User {user_id} already existed; record confirmed.",
                created=False,
            )
        else:
            conn.execute(
                "INSERT INTO users (user_id, email, plan) VALUES (?,?,?)",
                (user_id, email, plan),
            )
            return ToolResult(
                success=True,
                message=f"User {user_id} created with plan={plan}.",
                created=True,
            )

import hashlib
import json
import sqlite3
from datetime import datetime

def send_alert_idempotent(
    db_path: str,
    recipient: str,
    subject: str,
    body: str,
    channel: str = "email",
) -> dict:
    """
    Send an alert exactly once. Safe to call multiple times with the same args.
    Returns {sent: True, idempotency_key: '...', duplicate: bool}.
    """
    # Deterministic idempotency key from content + recipient + channel
    content = json.dumps({"to": recipient, "subject": subject, "channel": channel},
                          sort_keys=True)
    idem_key = hashlib.sha256(content.encode()).hexdigest()[:16]

    with sqlite3.connect(db_path) as conn:
        conn.execute("""
            CREATE TABLE IF NOT EXISTS sent_alerts (
                idem_key TEXT PRIMARY KEY,
                sent_at TEXT NOT NULL,
                recipient TEXT,
                subject TEXT
            )
        """)
        existing = conn.execute(
            "SELECT sent_at FROM sent_alerts WHERE idem_key=?", (idem_key,)
        ).fetchone()

        if existing:
            return {
                "sent": True,
                "idempotency_key": idem_key,
                "duplicate": True,
                "originally_sent_at": existing[0],
            }

        # First send
        _do_send(recipient, subject, body, channel)  # replace with real send
        conn.execute(
            "INSERT INTO sent_alerts VALUES (?,?,?,?)",
            (idem_key, datetime.utcnow().isoformat(), recipient, subject),
        )
        return {
            "sent": True,
            "idempotency_key": idem_key,
            "duplicate": False,
        }

def _do_send(recipient, subject, body, channel):
    """Placeholder — replace with real email/Slack send."""
    print(f"[{channel.upper()}] → {recipient}: {subject}")

import crypto from "crypto";

interface ToolCall {
  name: string;
  args: Record<string, unknown>;
}

interface CachedResult {
  result: unknown;
  executedAt: string;
}

class IdempotentToolGateway {
  private cache = new Map<string, CachedResult>();
  private readonly sessionId: string;

  constructor(sessionId: string) {
    this.sessionId = sessionId;
  }

  private idempotencyKey(call: ToolCall): string {
    // Drop volatile args that would prevent cache hits on retry
    const { request_id, timestamp, trace_id, ...stable } = call.args;
    const content = JSON.stringify({
      session: this.sessionId,
      tool: call.name,
      args: Object.fromEntries(Object.entries(stable).sort()),
    });
    return crypto.createHash("sha256").update(content).digest("hex").slice(0, 16);
  }

  async execute(call: ToolCall): Promise<{ result: unknown; cached: boolean }> {
    const key = this.idempotencyKey(call);
    const cached = this.cache.get(key);

    if (cached) {
      console.log(`[IDEMPOTENT] Cache hit for ${call.name} (key=${key})`);
      return { result: cached.result, cached: true };
    }

    // First execution
    const result = await this.dispatchTool(call);
    this.cache.set(key, { result, executedAt: new Date().toISOString() });
    return { result, cached: false };
  }

  private async dispatchTool(call: ToolCall): Promise<unknown> {
    // Replace with real tool dispatch
    return `${call.name} executed with args: ${JSON.stringify(call.args)}`;
  }
}

// Usage
const gateway = new IdempotentToolGateway("session-abc123");

async function handleAgentToolCall(call: ToolCall) {
  const { result, cached } = await gateway.execute(call);
  if (cached) {
    return `[ALREADY DONE] ${result}`;  // model sees success, stops retrying
  }
  return result;
}

from runguard import LoopDetector, LoopDetectedError, BudgetExceededError

detector = LoopDetector(repeats=3, max_cycle_len=2)

def execute_tool_safe(
    gateway: "IdempotentToolGateway",
    call: ToolCall,
) -> str:
    """
    Execute a tool call with both idempotency (no duplicate effects)
    and loop detection (no runaway spend on duplicate calls).
    """
    # 1. Loop detection: trips if this pattern repeats 3+ times
    sig = f"{call['name']}:{sorted(call['args'].items())}"
    match = detector.record(sig)
    if match:
        raise LoopDetectedError(
            f"Loop detected for tool '{call['name']}': "
            f"pattern repeated {match.repeats}× — stopping agent.",
            match=match,
        )

    # 2. Idempotency: deduplicate the actual side effect
    result = gateway.execute_sync(call)
    return result["result"] if not result["cached"] else f"[CACHED] {result['result']}"