Budget and Loop Ceilings

Every experienced team has a story about the loop that ran all weekend. The guard is twenty lines of code and there is no defensible reason to ship an agent without it.

Four ceilings, one object

Steps alone are not enough — one step that retrieves a 200k-token document blows a token budget that fifteen normal steps would not. Track all four dimensions in one object, check before each iteration, and record consumption after each. Spend is the one non-engineers care about, so make it a first-class dimension rather than something inferred from tokens later.

@dataclass
class Budget:
    max_steps: int = 20
    max_tokens: int = 200_000
    max_seconds: float = 300
    max_cost_usd: float = 2.00

    steps: int = 0
    tokens: int = 0
    cost_usd: float = 0.0
    started: float = field(default_factory=time.monotonic)

    def exhausted(self) -> str | None:
        if self.steps >= self.max_steps:
            return f"step limit ({self.max_steps})"
        if self.tokens >= self.max_tokens:
            return f"token limit ({self.max_tokens:,})"
        if time.monotonic() - self.started >= self.max_seconds:
            return f"time limit ({self.max_seconds}s)"
        if self.cost_usd >= self.max_cost_usd:
            return f"cost limit (${self.max_cost_usd:.2f})"
        return None

    def charge(self, usage: Usage) -> None:
        self.steps += 1
        self.tokens += usage.input_tokens + usage.output_tokens
        self.cost_usd += price(usage)

Stop gracefully, not abruptly

When a budget is exhausted, give the model one final turn to summarize what it accomplished — with tools disabled so it cannot start new work. A partial answer with a stated limitation is far more useful to a caller than a raised exception, and it costs one cheap call.

async def run(task: str, budget: Budget) -> Result:
    messages = [Message(role="user", content=task)]

    while True:
        if (reason := budget.exhausted()):
            return await wrap_up(messages, reason, budget)

        response = await model(messages, tools=TOOLS)
        budget.charge(response.usage)
        messages.append(response.message)

        if not response.tool_calls:
            return Result(status="complete", output=response.text, budget=budget)

        messages.extend(await execute_batch(response.tool_calls))


async def wrap_up(messages: list[Message], reason: str, budget: Budget) -> Result:
    metrics.increment("agent.budget_exhausted", tags={"reason": reason.split(" ")[0]})
    messages.append(Message(role="user", content=(
        f"You have reached the {reason}. Stop working. Summarize what you "
        f"established, what remains unfinished, and the single best next step."
    )))
    final = await model(messages, tools=[])   # no tools: it cannot start anything new
    return Result(status="incomplete", output=final.text, stopped_by=reason, budget=budget)

Detecting progress-free loops

Some loops stay comfortably inside every ceiling while accomplishing nothing — the same search, the same failing tool call, the same reasoning restated. Fingerprint each step and stop when a fingerprint repeats too often. This catches the pathology much earlier than a step ceiling, and the earlier you catch it the less it cost.

def step_fingerprint(response: Response) -> str:
    calls = sorted((c.name, canonical(c.arguments)) for c in response.tool_calls)
    return hashlib.sha1(repr(calls).encode()).hexdigest()[:12]


class LoopDetector:
    def __init__(self, repeat_limit: int = 3):
        self.seen, self.limit = Counter(), repeat_limit

    def check(self, response: Response) -> str | None:
        if not response.tool_calls:
            return None
        fp = step_fingerprint(response)
        self.seen[fp] += 1
        if self.seen[fp] >= self.limit:
            return f"repeated the same tool calls {self.limit} times without progress"
        return None

• Enforce steps, tokens, wall clock, and spend — any one alone is insufficient
• Share the budget across the whole agent tree, not per agent
• On exhaustion, take one final tool-free turn to summarize; return a partial result
• Fingerprint steps to catch progress-free loops well before the ceilings fire
• Emit budget consumption as telemetry — it is your best cost-attribution signal