Stopping a $946 Token Leak with Real-Time Agent Guardrails

We had an autonomous agent running a classic THINK → ACT → OBSERVE loop over Claude on AWS Bedrock. It worked. It also, every couple of weeks, quietly set fire to about $946 in tokens.

The failure mode was always the same shape. The agent would call a tool, get a result it didn't quite know what to do with, and call the same tool again with almost the same arguments. Then again. Then again. Each pass appended the prior result back into context, so the context window grew on every iteration. Left unsupervised, a single run could cross a million tokens before either the model gave up or a hard limit tripped. Multiply that by a few stuck sessions a week and you get a four-figure invoice for the privilege of watching an agent spin in place.

The naive fix — a hard iteration cap — doesn't actually solve this. A cap of 20 still lets a degenerate loop chew through enormous context before it stops, and it kills legitimately long, productive runs. We needed something that understood why the loop was pathological, not just that it was long.

The mechanism: hooks around every tool call

The core idea is a wrapper, HookedToolNode, that sits between the agent and its tools, plus a HookRunner that orchestrates a chain of pre- and post-tool hooks. Every tool invocation passes through it. A hook can inspect the call before it runs and the result after, and return one of four actions:

from enum import Enum
 
class HookAction(str, Enum):
    PASS = "pass"        # let the call proceed untouched
    RESOLVE = "resolve"  # short-circuit: return this result, skip the tool
    REWRITE = "rewrite"  # mutate args (or result) and continue
    OBSERVE = "observe"  # record telemetry only, change nothing

That four-action model is the whole design. PASS is the default. OBSERVE is for metrics and tracing with zero behavioral effect. REWRITE lets a hook normalize or trim arguments. RESOLVE is the powerful one — it lets a hook answer on behalf of the tool, which is exactly what you want when you've detected a loop and need to inject a corrective signal instead of running the redundant call.

The hook interface itself is deliberately small:

from dataclasses import dataclass
from typing import Any, Optional
 
@dataclass
class HookResult:
    action: HookAction
    payload: Optional[Any] = None   # rewritten args, resolved result, or note
    reason: str = ""
 
class Hook:
    name: str
 
    def pre(self, tool_name: str, args: dict, ctx: "RunContext") -> HookResult:
        return HookResult(HookAction.PASS)
 
    def post(self, tool_name: str, args: dict, result: Any,
             ctx: "RunContext") -> HookResult:
        return HookResult(HookAction.PASS)

HookRunner walks the registered hooks in order. The first hook to return anything other than PASS/OBSERVE wins and the runner short-circuits — so ordering is a real design lever. Cheap, high-confidence hooks (loop detection) go first; expensive or advisory ones go last.

Detecting the pathological loop

The detector doesn't care about iteration count. It cares about repetition. On every pre-hook it computes a fingerprint of the call — the tool name plus its normalized arguments — and counts how many times that fingerprint has appeared in a recent sliding window.

Normalization is what makes this robust. Agents rarely repeat byte-for-byte; they'll reorder keys, flip whitespace, or nudge a value. So before hashing we sort keys, lowercase and strip string values, and round floats. Near-identical calls collapse to the same fingerprint.

import hashlib, json
from collections import deque
 
def fingerprint(tool_name: str, args: dict) -> str:
    def norm(v):
        if isinstance(v, str):
            return " ".join(v.lower().split())
        if isinstance(v, float):
            return round(v, 4)
        if isinstance(v, dict):
            return {k: norm(v[k]) for k in sorted(v)}
        if isinstance(v, list):
            return [norm(x) for x in v]
        return v
    canon = json.dumps({"t": tool_name, "a": norm(args)}, sort_keys=True)
    return hashlib.sha1(canon.encode()).hexdigest()
 
class LoopGuard(Hook):
    name = "loop_guard"
 
    def __init__(self, window: int = 8, threshold: int = 3):
        self.window = deque(maxlen=window)
        self.threshold = threshold
 
    def pre(self, tool_name, args, ctx) -> HookResult:
        fp = fingerprint(tool_name, args)
        self.window.append(fp)
        repeats = sum(1 for x in self.window if x == fp)
        if repeats >= self.threshold:
            return HookResult(
                action=HookAction.RESOLVE,
                payload=(
                    f"Loop detected: `{tool_name}` called with near-identical "
                    f"arguments {repeats}x in the last {len(self.window)} steps. "
                    "Stop repeating this call. Re-read the previous result, change "
                    "your approach, or report that you are blocked."
                ),
                reason="loop_guard.repeat_threshold",
            )
        return HookResult(HookAction.PASS)

When the threshold trips, the guard RESOLVEs. The tool never runs again. Instead the agent receives a blunt, structured message back in the observation slot telling it exactly what it's doing wrong and what to do instead. In practice the model reads that, breaks the pattern, and either changes tack or cleanly reports it's stuck — both of which are vastly cheaper than another lap around the loop.

The cheapest token is the one you never send. Guardrails aren't about making the agent smarter; they're about refusing to pay for the same mistake twice.

What it cost to build, and what it saved

The whole layer is about 700 lines: the wrapper, the runner, the action model, the loop detector, and a couple of advisory hooks for context-size budgeting. It ships with 86 tests covering hook ordering, the short-circuit semantics, fingerprint normalization edge cases, and the window/threshold math. Zero regressions against the existing agent behavior, because PASS is the default and untriggered runs are byte-identical to before.

The results that mattered:

• The recurring ~$946 / two-week token waste went to zero. Degenerate loops now terminate in single-digit iterations instead of running until a context limit.
• We stopped seeing 1M+ token context blowouts entirely. The loop guard catches the runaway long before context can balloon.
• Because the resolve message is structured and specific, the agent's recovery improved too — it more often produces a useful "I'm blocked because X" instead of failing opaquely.

The broader lesson: in agentic systems, the expensive failures are rarely single bad calls. They're patterns — the same wrong move, repeated, each repetition compounding the cost through context growth. You can't catch that with per-call validation. You need a thin orchestration layer with memory of recent behavior and the authority to intervene mid-loop. The four-action model gave us exactly enough authority, and not one bit more.