The Definitive AI Agent Cost Control Pattern Reference

We have now published cost control guides for 35 AI agent frameworks — from LangGraph and CrewAI through AWS Bedrock, Azure AI, Google ADK, Vertex AI, IBM WatsonX, Salesforce Agentforce, Microsoft Copilot Studio, Dify, Flowise, n8n, Ollama, llama.cpp, and three dozen others. Every guide covers the same four failure modes. Every guide builds guard implementations from the same four detection algorithms. The frameworks differ in syntax, execution model, and state management API; the underlying failure patterns do not.

This post is the reference that should have existed from the beginning: the four universal failure modes described precisely, the detection algorithms explained from first principles, portable Python and TypeScript implementations you can adapt to any framework, a decision tree for choosing what to guard first, and a complete index of every framework-specific guide in this series. If you are starting from scratch on cost control for your agent, read this page first and then follow the link to your framework.

The four universal failure modes

Every runaway agent incident we have documented falls into exactly one of four categories. The taxonomy holds regardless of whether the agent is running inside a Python orchestration library, a cloud-managed platform, a visual no-code builder, or a TypeScript SDK.

Failure mode 1 — Tool call spiral

The agent calls the same tool (or the same set of tools in the same sequence) with semantically identical arguments across consecutive iterations. The model is stuck: it receives a result, determines it is insufficient, and re-issues the same query with minor surface variation. The tool call count climbs while no meaningful work advances. Each iteration costs one LLM reasoning step plus one or more tool invocations.

Canonical signal: Jaccard similarity ≥ 0.72 between the argument sets of the most recent tool call and any of the preceding three calls to the same tool. A Jaccard threshold of 0.72 reliably catches spirals while allowing legitimate narrow refinement (a search query that zooms in on a topic reads as low-similarity even though the words partially overlap).

Why the model gets stuck here: The model is trained to keep trying when a tool returns an unsatisfying result. It has no mechanism to recognize that it already tried this and got the same answer. Without an external guard watching argument fingerprints, it will re-issue indefinitely.

Failure mode 2 — Context window accumulation

The agent's conversation history grows with each iteration — appending the user turn, the model's reasoning, tool arguments, and tool results into the context passed to the next LLM call. Eventually the accumulated context exceeds the model's effective context window. At that point the model silently truncates the oldest content and begins re-researching information it already worked through. Each new iteration makes the problem worse: more tokens are added, more old context is cut, more re-research is triggered.

Canonical signal: Estimated token count of the full context reaches 70% of the model's context limit (soft warn) or 85% (hard stop). Token estimation via len(text.split()) * 1.35 — words times 1.35 is an accurate approximation for mixed code/prose content across GPT-4, Claude, and Gemini tokenizers without requiring a tokenizer import.

Why this is silent: The model does not raise an error when truncation happens. It simply begins its reasoning with fewer facts than it had a turn ago. The output quality degrades visibly to a human reader but the agent loop continues without any signal that something is wrong.

Failure mode 3 — Retry cascade multiplication

The agent's framework has retry logic for failed tool calls (usually 3–5 retries with exponential backoff). The tool being called is a remote service that is down, rate-limited, or returning errors. The framework's retry logic fires once per agent step; the agent loop fires for every model turn that includes that tool call; the external service's own SDK may have its own retry layer. The result is multiplicative: 3 framework retries × 5 agent loop iterations × 2 SDK retries = 30 calls to a downed system in a single run. Each call waits out the full backoff timer, turning a 20-second agent run into a 20-minute bill-accruing stall.

Canonical signal: Two or more consecutive failures of the same tool within a single run. On the second consecutive failure, open the circuit breaker for that tool: return a synthetic "service unavailable — do not retry this session" result to the model rather than calling the external service again. Reset after the run completes.

Why per-run state matters: Retry logic at the SDK or framework level has no visibility into how many times the model has already requested the same tool. The circuit breaker must be maintained in run-scoped state that the framework's own retry logic cannot reach.

Failure mode 4 — Budget breach

The agent run consumes more tokens than the task warrants because it is in one of the above three failure modes, because the context it was handed was unexpectedly large, or because the model chose an unusually token-expensive reasoning path. The framework has no mechanism to halt a run once spending exceeds a threshold — it simply continues until the model returns a final answer or a hard iteration ceiling is hit.

Canonical signal: Cumulative token spend for the run exceeds a per-run budget. The budget is set conservatively — 2× the p95 token cost of a successful run for this task type. Track input tokens and output tokens separately because output tokens cost 3–5× more per token on most models.

Why soft limits fail: Logging a warning when spend is high does not stop the run. The model does not read your log stream. The guard must inject a hard stop into the agent loop — either by raising an exception that the framework catches, or by returning a budget-exceeded signal in the tool result that the model is prompted to treat as terminal.

Pattern 1 — Tool call spiral detection

The spiral guard maintains a rolling window of the last N argument sets for each tool, converts each set to a normalized token bag, and computes Jaccard similarity between the current call and each prior call in the window. If any pair exceeds the threshold, the guard fires.

Python implementation

import json
import re

def _tokenize(text: str) -> set[str]:
    return set(re.findall(r'\w+', text.lower()))

class SpiralGuard:
    def __init__(self, window: int = 4, threshold: float = 0.72):
        self.window = window
        self.threshold = threshold
        self._history: dict[str, list[set[str]]] = {}

    def check(self, tool_name: str, args: dict) -> bool:
        """Returns True if a spiral is detected (caller should abort)."""
        fingerprint = _tokenize(json.dumps(args, sort_keys=True))
        history = self._history.setdefault(tool_name, [])

        for prior in history:
            if len(fingerprint | prior) == 0:
                continue
            jaccard = len(fingerprint & prior) / len(fingerprint | prior)
            if jaccard >= self.threshold:
                return True

        history.append(fingerprint)
        if len(history) > self.window:
            history.pop(0)
        return False

# Usage in any Python agent framework:
spiral = SpiralGuard()

def guarded_tool_call(tool_name: str, args: dict):
    if spiral.check(tool_name, args):
        return {"error": "spiral_detected", "message": "Repeated tool call pattern detected. Stop and report current findings."}
    return original_tool_call(tool_name, args)

TypeScript implementation

function tokenize(text: string): Set<string> {
  return new Set(text.toLowerCase().match(/\w+/g) ?? []);
}

function jaccard(a: Set<string>, b: Set<string>): number {
  const union = new Set([...a, ...b]);
  if (union.size === 0) return 0;
  const intersection = [...a].filter(x => b.has(x));
  return intersection.length / union.size;
}

class SpiralGuard {
  private history = new Map<string, Set<string>[]>();
  constructor(private window = 4, private threshold = 0.72) {}

  check(toolName: string, args: Record<string, unknown>): boolean {
    const fingerprint = tokenize(JSON.stringify(args));
    const history = this.history.get(toolName) ?? [];

    for (const prior of history) {
      if (jaccard(fingerprint, prior) >= this.threshold) return true;
    }

    history.push(fingerprint);
    if (history.length > this.window) history.shift();
    this.history.set(toolName, history);
    return false;
  }
}

// Usage:
const spiral = new SpiralGuard();
function guardedToolCall(toolName: string, args: Record<string, unknown>) {
  if (spiral.check(toolName, args)) {
    return { error: 'spiral_detected', message: 'Repeated pattern. Stop and summarise findings.' };
  }
  return originalToolCall(toolName, args);
}

Pattern 2 — Context window guard

The context guard estimates the token count of the full context before each LLM call. It injects a summary directive at the soft threshold (70%) and halts the run at the hard threshold (85%), returning the best answer available from accumulated state so far.

Python implementation

class ContextGuard:
    def __init__(self, context_limit: int, soft_pct: float = 0.70, hard_pct: float = 0.85):
        self.soft_limit = int(context_limit * soft_pct)
        self.hard_limit = int(context_limit * hard_pct)

    def estimate_tokens(self, text: str) -> int:
        return int(len(text.split()) * 1.35)

    def check(self, context: str) -> tuple[str, int]:
        """Returns ('ok'|'warn'|'halt', estimated_tokens)."""
        tokens = self.estimate_tokens(context)
        if tokens >= self.hard_limit:
            return 'halt', tokens
        if tokens >= self.soft_limit:
            return 'warn', tokens
        return 'ok', tokens

# Usage: wrap your LLM call function
ctx_guard = ContextGuard(context_limit=128_000)  # adjust to your model

def guarded_llm_call(messages: list[dict]) -> str:
    full_text = " ".join(m.get("content", "") for m in messages)
    status, tokens = ctx_guard.check(full_text)

    if status == 'halt':
        return "Context limit approaching. Stopping to avoid truncation. Summarise findings now."
    if status == 'warn':
        messages = messages + [{
            "role": "system",
            "content": f"Context is at {tokens} estimated tokens. Consolidate and produce final answer."
        }]
    return llm_call(messages)

TypeScript implementation

type ContextStatus = 'ok' | 'warn' | 'halt';

class ContextGuard {
  private softLimit: number;
  private hardLimit: number;

  constructor(contextLimit: number, softPct = 0.70, hardPct = 0.85) {
    this.softLimit = Math.floor(contextLimit * softPct);
    this.hardLimit = Math.floor(contextLimit * hardPct);
  }

  estimateTokens(text: string): number {
    return Math.floor(text.split(/\s+/).length * 1.35);
  }

  check(context: string): [ContextStatus, number] {
    const tokens = this.estimateTokens(context);
    if (tokens >= this.hardLimit) return ['halt', tokens];
    if (tokens >= this.softLimit) return ['warn', tokens];
    return ['ok', tokens];
  }
}

const ctxGuard = new ContextGuard(128_000);

async function guardedLlmCall(messages: Array<{role: string; content: string}>) {
  const fullText = messages.map(m => m.content).join(' ');
  const [status, tokens] = ctxGuard.check(fullText);

  if (status === 'halt') {
    return 'Context limit approaching. Stopping to preserve quality. Report findings.';
  }
  if (status === 'warn') {
    messages = [...messages, {
      role: 'system',
      content: `Context at ~${tokens} tokens. Consolidate and produce final answer.`
    }];
  }
  return llmCall(messages);
}

Pattern 3 — Retry circuit breaker

The circuit breaker tracks consecutive failures per tool within a run. On the second consecutive failure it opens the circuit for that tool: subsequent calls return a synthetic "service unavailable" response without hitting the external service. The circuit resets when the run ends.

Python implementation

class RetryCircuitBreaker:
    def __init__(self, failure_threshold: int = 2):
        self.threshold = failure_threshold
        self._consecutive: dict[str, int] = {}
        self._open: set[str] = set()

    def record_failure(self, tool_name: str) -> bool:
        """Records a failure. Returns True if circuit is now open."""
        count = self._consecutive.get(tool_name, 0) + 1
        self._consecutive[tool_name] = count
        if count >= self.threshold:
            self._open.add(tool_name)
            return True
        return False

    def record_success(self, tool_name: str) -> None:
        self._consecutive[tool_name] = 0
        self._open.discard(tool_name)

    def is_open(self, tool_name: str) -> bool:
        return tool_name in self._open

# Usage:
breaker = RetryCircuitBreaker()

def guarded_tool_call(tool_name: str, args: dict):
    if breaker.is_open(tool_name):
        return {
            "error": "circuit_open",
            "message": f"{tool_name} is unavailable. Do not retry this session. Continue without it."
        }
    try:
        result = original_tool_call(tool_name, args)
        breaker.record_success(tool_name)
        return result
    except Exception as e:
        if breaker.record_failure(tool_name):
            return {"error": "circuit_open", "message": f"{tool_name} failed twice. Skipping for this run."}
        raise

TypeScript implementation

class RetryCircuitBreaker {
  private consecutive = new Map<string, number>();
  private open = new Set<string>();

  constructor(private threshold = 2) {}

  recordFailure(toolName: string): boolean {
    const count = (this.consecutive.get(toolName) ?? 0) + 1;
    this.consecutive.set(toolName, count);
    if (count >= this.threshold) {
      this.open.add(toolName);
      return true;
    }
    return false;
  }

  recordSuccess(toolName: string): void {
    this.consecutive.set(toolName, 0);
    this.open.delete(toolName);
  }

  isOpen(toolName: string): boolean {
    return this.open.has(toolName);
  }
}

const breaker = new RetryCircuitBreaker();

async function guardedToolCall(toolName: string, args: Record<string, unknown>) {
  if (breaker.isOpen(toolName)) {
    return { error: 'circuit_open', message: `${toolName} unavailable. Do not retry. Continue without it.` };
  }
  try {
    const result = await originalToolCall(toolName, args);
    breaker.recordSuccess(toolName);
    return result;
  } catch (err) {
    if (breaker.recordFailure(toolName)) {
      return { error: 'circuit_open', message: `${toolName} failed twice. Skipping this run.` };
    }
    throw err;
  }
}

Pattern 4 — Budget enforcement

The budget guard accumulates token spend across all LLM calls in a run (input and output counted separately, since output tokens cost more). When cumulative spend exceeds the per-run budget ceiling, the next LLM call is intercepted and a final-answer directive is injected or the run is halted outright.

Python implementation

class BudgetGuard:
    """
    Tracks per-run token spend. Costs in USD per million tokens.
    Default pricing is approximate Claude Sonnet / GPT-4o tier.
    """
    def __init__(
        self,
        max_usd: float,
        input_cost_per_m: float = 3.0,
        output_cost_per_m: float = 15.0,
    ):
        self.max_usd = max_usd
        self.input_cpp = input_cost_per_m / 1_000_000
        self.output_cpp = output_cost_per_m / 1_000_000
        self.spent = 0.0

    def record(self, input_tokens: int, output_tokens: int) -> bool:
        """Records spend. Returns True if budget is exceeded."""
        self.spent += input_tokens * self.input_cpp + output_tokens * self.output_cpp
        return self.spent >= self.max_usd

    @property
    def remaining(self) -> float:
        return max(0.0, self.max_usd - self.spent)

# Usage:
budget = BudgetGuard(max_usd=0.50)  # $0.50 per run ceiling

def guarded_llm_call(messages: list[dict]) -> str:
    if budget.remaining == 0:
        return "Budget exceeded. Stopping run. Report best available answer."
    response = llm_call(messages)
    exceeded = budget.record(response.usage.input_tokens, response.usage.output_tokens)
    if exceeded:
        return f"{response.content}\n\n[Budget ceiling reached at ${budget.spent:.4f}. No further calls.]"
    return response.content

TypeScript implementation

class BudgetGuard {
  private spent = 0;

  constructor(
    private maxUsd: number,
    private inputCostPerM = 3.0,
    private outputCostPerM = 15.0,
  ) {}

  record(inputTokens: number, outputTokens: number): boolean {
    this.spent +=
      (inputTokens * this.inputCostPerM) / 1_000_000 +
      (outputTokens * this.outputCostPerM) / 1_000_000;
    return this.spent >= this.maxUsd;
  }

  get remaining(): number {
    return Math.max(0, this.maxUsd - this.spent);
  }
}

const budget = new BudgetGuard(0.50);

async function guardedLlmCall(messages: Array<{role: string; content: string}>) {
  if (budget.remaining === 0) {
    return 'Budget exceeded. Stopping. Report best available answer.';
  }
  const response = await llmCall(messages);
  const exceeded = budget.record(response.usage.inputTokens, response.usage.outputTokens);
  if (exceeded) {
    return `${response.content}\n\n[Budget ceiling reached. No further calls.]`;
  }
  return response.content;
}

Composing the full guard stack

The four guards are independent and composable. The recommended ordering for a single agent run is:

1. Budget check (pre-call) — if the run is already over budget, return the budget-exceeded message immediately without calling the LLM.
2. Circuit breaker check (pre-tool) — if the tool circuit is open, return the synthetic unavailability message without calling the tool.
3. Spiral check (pre-tool) — if the argument fingerprint matches a prior call in the window, return the spiral-detected message without calling the tool.
4. Context check (pre-LLM) — estimate token count before each LLM call; inject the soft-warn directive or return the halt message at the hard threshold.
5. Budget record (post-LLM) — update cumulative spend with actual token counts from the response.

In Python this composes naturally as a thin wrapper around your framework's tool execution hook and LLM call function. In TypeScript it wraps the SDK's generate or chat function and the tool execution callback. All four guards operate on state local to the run — there is no shared mutable state between runs, which means they are safe to run in concurrent multi-tenant deployments as long as the guard instances are created per-run rather than shared.

Choosing what to guard first

If you are adding guards to an existing agent for the first time and need to prioritize, use this decision tree:

Your agent primarily calls	Start with	Why
Search tools or RAG retrieval	Spiral guard	Search tools with near-identical queries are the single most common runaway pattern in the wild.
Long multi-turn conversations	Context guard	History accumulation is the dominant cost driver in conversational agents with no session reset.
External HTTP APIs or databases	Retry circuit breaker	Transient failures in downstream services are the fastest path to a multi-minute stall compounding with framework retries.
Long-horizon autonomous tasks	Budget guard	Any task with an open-ended horizon needs a hard dollar ceiling before it runs in production.

In production, deploy all four. The guards are lightweight enough that the overhead is immaterial — the spiral guard adds a few microseconds per tool call, the context guard a fraction of a millisecond per LLM call. The cost of not having them, documented repeatedly across every framework guide in this series, starts at $50 per incident and scales with how long the loop runs before a human notices.

Framework-specific implementation guides

The following guides implement all four patterns against each framework's specific execution model — its tool call hooks, context management API, retry configuration, and state persistence primitives. Start with the guide for your framework, then return here if you need to adapt the patterns for a different deployment target.

Open-source Python agent frameworks

TypeScript / JavaScript frameworks

Cloud platform agents

Local / self-hosted inference

Enterprise / low-code platforms

Foundational guides

Using RunGuard as a managed alternative

The patterns above require you to instrument every agent framework you deploy, maintain guard state across framework upgrades, and monitor guard trips yourself. RunGuard packages all four guards as a managed API so the guard logic lives outside your application code.

The integration point is a single HTTP call before each tool execution:

POST https://api.runguard.dev/v1/guard
{
  "app_id": "your-app-id",
  "session_id": "run-abc123",
  "tool_name": "web_search",
  "tool_args_hash": "sha256-of-normalized-args",
  "call_count": 4,
  "context_tokens": 89420,
  "budget_spent_usd": 0.23
}

// Response if safe:
{ "allow": true }

// Response if spiral detected:
{ "allow": false, "reason": "spiral", "message": "Repeated search pattern. Report findings." }

// Response if budget exceeded:
{ "allow": false, "reason": "budget", "message": "Per-run ceiling reached. Conclude run." }

The dashboard at runguard.dev/dashboard shows a 30-day chart of guard trips per app, broken down by failure mode and tool. The circuit breaker and budget guard are configurable per app without a code deploy. The 14-day free trial includes full access to all guard types and the dashboard.