Testing agent loops: budgets, termination and failure recovery

Learn in your browser · Chapter 01 of 06

A model can propose an action. The harness decides whether that action may run, records the outcome and eventually stops. Test that machinery with predictable inputs before asking whether the model gives useful answers.

Imagine a fictional opening-hours assistant. Its only tool reads a tiny in-memory dictionary. It may request a lookup or return a final answer. It cannot run commands, send messages or modify files. This deliberately small action space lets us inspect every side effect. Replacing the dictionary with a real service would require additional authorisation, validation and integration tests.

Write a contract that someone else could test without reading the implementation. An attempted model call consumes one step, even when it raises an exception. Every attempted tool call consumes one tool allowance, including a failed attempt and a retry. The deadline is checked before work and after a successful return. A late value is discarded. Completion means that a valid finish action was received within the deadline; it does not establish that its answer is true.

Scroll sideways to see every column.

These are independent tests of different limits. A run with three steps and two tool attempts may stop at the tool limit before the step limit. Test each budget with the other budgets high enough not to mask it. Also test boundary equality: if the deadline is 1.0 and the clock reads 1.0, our policy says the result is late.

Learn in your browser · Chapter 02 of 06

Create a new local folder for this lab and use Python 3.12 or later. The worked example uses only the standard library. Save each named file as UTF-8; copy the selectable web code if a PDF reader changes indentation.

Create harness.py. Put the following imports and support types at its top. A frozen Result keeps the top-level report fields fixed after return. Its trace is a tuple of event labels; it deliberately excludes raw prompts, tool inputs and exception messages. This is useful minimal evidence, not a durable audit system.

from dataclasses import dataclass
from math import isfinite
from time import monotonic


class TemporaryToolError(Exception):
    pass


@dataclass(frozen=True)
class Result:
    status: str
    steps: int
    tool_calls: int
    trace: tuple
    answer: str = ""


def positive_int(value):
    return type(value) is int and value > 0

Append the next block to harness.py. It starts run(), validates configuration and implements the model half of the loop. Do not execute the file until you have appended chapter 3, which completes the indented while body. Python treats True as an integer in many contexts; the exact type check deliberately rejects boolean values used accidentally as budgets.

def run(model, tool, *, max_steps=3, max_tool_calls=2,
        seconds=1.0, clock=monotonic):
    if not positive_int(max_steps):
        raise ValueError("max_steps must be a positive integer")
    if not positive_int(max_tool_calls):
        raise ValueError("max_tool_calls must be positive")
    if (type(seconds) not in (int, float)
            or not isfinite(seconds) or seconds <= 0):
        raise ValueError("seconds must be positive and finite")

    deadline = clock() + seconds
    history, trace = [], []
    steps = calls = 0

    def expired():
        return clock() >= deadline

    def finish(status, answer=""):
        return Result(status, steps, calls, tuple(trace), answer)

    while steps < max_steps:
        if expired():
            return finish("deadline")
        steps += 1  # An attempted call consumes a step.
        trace.append("model")
        try:
            action = model(tuple(history))
        except Exception:
            return finish("model_error")
        if expired():
            return finish("deadline")
        if not isinstance(action, dict):
            return finish("invalid_action")
        kind = action.get("kind")
        if kind == "finish":
            answer = action.get("answer")
            if type(answer) is not str or not answer.strip():
                return finish("invalid_action")
            if len(answer) > 200:
                return finish("invalid_action")
            return finish("finished", answer)
        if kind != "lookup":
            return finish("invalid_action")
        key = action.get("key")
        if type(key) is not str or not 1 <= len(key) <= 80:
            return finish("invalid_action")

The model receives an immutable tuple of previous tool-result strings. Passing dependencies as arguments is dependency injection: the caller can supply controlled substitutes without rewriting the loop. The clock defaults to time.monotonic for elapsed-time comparisons. Its absolute value has no calendar meaning, and only differences are relevant here. In tests we supply a clock whose value changes only when the test says so.

The action contract is intentionally narrow. Unknown kinds, blank final answers, long answers and invalid lookup keys return invalid_action. Extra dictionary fields are ignored and never forwarded. This is not a general schema validator. A production adapter should validate the full protocol, permissions and resource limits before exposing powerful tools. A valid final answer can still be nonsense: structural validity and answer quality are different questions.

Learn in your browser · Chapter 03 of 06

Append this block directly after chapter 2 in harness.py. Preserve its leading spaces: the for loop belongs inside the while loop, and the final return belongs outside it but inside run().

        # Two attempts at most, including the first attempt.
        for attempt in range(2):
            if expired():
                return finish("deadline")
            if calls >= max_tool_calls:
                return finish("tool_limit")
            calls += 1
            trace.append("lookup")
            try:
                value = tool(key)
            except TemporaryToolError:
                trace.append("temporary_error")
                if expired():
                    return finish("deadline")
                if attempt == 1:
                    return finish("tool_error")
                continue
            except Exception:
                return finish("tool_error")
            if expired():
                return finish("deadline")
            if type(value) is not str or len(value) > 200:
                return finish("invalid_result")
            history.append(value)
            break
    return finish("step_limit")

Only TemporaryToolError is retried, and only once. The original attempt and its retry both consume the tool budget. Other exceptions stop immediately. This is an explicit policy for a fictional read-only lookup. It is not permission to retry payments, deletions or other actions with consequences. If a service times out after performing an action, repeating it may duplicate the action. Such operations need their own idempotency and reconciliation design.

The word cooperative matters. If a synchronous tool never returns, this function cannot regain control to check its clock. A fake deadline test proves rejection of a late result after the callback returns; it does not prove pre-emption. Real adapters need transport timeouts and, where necessary, cancellable work or an isolated worker that can be stopped. Even cancellation does not imply that a remote action was rolled back.

Failure status precedence is also part of this example. A normal model or tool return is followed by a deadline check. A generic dependency exception reports model_error or tool_error; a temporary tool exception checks the deadline before attempting recovery. Document that policy rather than promising that every run which used too much wall-clock time always reports deadline. The trace helps explain the path that actually occurred.

Count budgets are not a money meter. A model request may consume different token counts and prices from another request. This lab enforces attempts only. A service that promises a spending bound must reserve and reconcile provider usage, account for concurrent requests and use enforceable service limits. Logging an estimated cost after a response cannot prevent that response from already incurring a charge.

Learn in your browser · Chapter 04 of 06

A scripted model returns a known sequence of actions. That lets you test the harness without depending on sampling, network availability or a provider bill. Save the next block in fixtures.py beside harness.py.

from harness import run, TemporaryToolError


class ScriptedModel:
    def __init__(self, actions):
        self.actions = iter(actions)
        self.seen = []

    def __call__(self, history):
        self.seen.append(history)
        return next(self.actions)


class FakeClock:
    def __init__(self):
        self.now = 0.0

    def __call__(self):
        return self.now


def lookup(key):
    return {"opening": "09:00"}[key]


if __name__ == "__main__":
    model = ScriptedModel([
        {"kind": "lookup", "key": "opening"},
        {"kind": "finish", "answer": "Opens at 09:00."},
    ])
    result = run(model, lookup, clock=FakeClock())
    print(result.status, result.steps, result.tool_calls)
    print(result.answer)
    print(result.trace)

Run python fixtures.py in the lab folder. You should see the output below. The trace describes two model calls with one lookup between them. The answer is fictional fixture text, not a statement about a real organisation. A scripted sequence that is exhausted raises StopIteration; the harness reports model_error rather than silently inventing another response.

finished 2 1
Opens at 09:00.
('model', 'lookup', 'model')

FakeClock begins at zero and stays there until a test changes now. It does not sleep. Advancing it inside a tool simulates elapsed time between a call and its return. Keep model decisions, data fixtures and time independent: a failure should tell you whether a policy boundary broke, not whether the computer happened to run slowly that morning.

A replay is only as informative as its cases. Replaying one success trace cannot establish robustness against malformed requests, repeated actions or a different protocol. For a real integration, create a sanitised fixture from the documented adapter boundary. Retain event order, schema version and relevant outcomes. Exclude personal data and credentials; use invented records wherever possible. Replaying internal model reasoning is unnecessary for testing this contract.

Learn in your browser · Chapter 05 of 06

Save the next three code blocks, in order, in test_harness.py. The later blocks continue the LoopTests class before the final runner. Run python -m unittest -v test_harness from the same folder.

import unittest
from unittest.mock import Mock
from harness import run, TemporaryToolError
from fixtures import ScriptedModel, FakeClock, lookup


class LoopTests(unittest.TestCase):
    def test_normal_completion(self):
        model = ScriptedModel([
            {"kind": "lookup", "key": "opening"},
            {"kind": "finish", "answer": "09:00"},
        ])
        tool = Mock(side_effect=lookup)
        r = run(model, tool, clock=FakeClock())
        self.assertEqual((r.status, r.steps, r.tool_calls),
                         ("finished", 2, 1))
        self.assertEqual(model.seen, [(), ("09:00",)])
        tool.assert_called_once_with("opening")

    def test_repetition_stops_at_step_limit(self):
        model = Mock(return_value={"kind": "lookup", "key": "x"})
        tool = Mock(return_value="same answer")
        r = run(model, tool, max_steps=3, max_tool_calls=9,
                clock=FakeClock())
        self.assertEqual(r.status, "step_limit")
        self.assertEqual(model.call_count, 3)
        self.assertEqual(tool.call_count, 3)

Continue the class with these two methods. Keep the four-space indentation before each def.

    def test_invalid_action_never_reaches_tool(self):
        tool = Mock()
        r = run(lambda _: {"kind": "erase"}, tool,
                clock=FakeClock())
        self.assertEqual(r.status, "invalid_action")
        tool.assert_not_called()

    def test_temporary_failure_retries_once(self):
        model = ScriptedModel([
            {"kind": "lookup", "key": "opening"},
            {"kind": "finish", "answer": "09:00"},
        ])
        tool = Mock(side_effect=[TemporaryToolError(), "09:00"])
        r = run(model, tool, clock=FakeClock())
        self.assertEqual(r.status, "finished")
        self.assertEqual(r.tool_calls, 2)
        self.assertEqual(tool.call_count, 2)

Append the remaining methods and the final test runner below.

    def test_retry_cannot_exceed_tool_budget(self):
        model = lambda _: {"kind": "lookup", "key": "opening"}
        tool = Mock(side_effect=TemporaryToolError())
        r = run(model, tool, max_tool_calls=1, clock=FakeClock())
        self.assertEqual(r.status, "tool_limit")
        self.assertEqual(tool.call_count, 1)

    def test_result_after_deadline_is_rejected(self):
        clock = FakeClock()

        def slow_tool(key):
            clock.now += 1.0
            return "09:00"

        model = Mock(return_value={"kind": "lookup", "key": "x"})
        r = run(model, slow_tool, seconds=1.0, clock=clock)
        self.assertEqual(r.status, "deadline")
        self.assertEqual(model.call_count, 1)
        self.assertEqual(r.tool_calls, 1)
        self.assertEqual(r.answer, "")


if __name__ == "__main__":
    unittest.main()

The six tests cover success, repeated model decisions, an invalid action, one successful recovery, an exhausted retry allowance and a deadline boundary. Mock records calls and can return a sequence of values or raise supplied exceptions through side_effect. Use assertions about the absence of calls as well as assertions about the final status. A loop that returns tool_limit after making an extra tool call is still wrong.

A passing suite should be challenged. In a disposable copy of harness.py, change the tool budget comparison from >= to >. The one-attempt retry test should fail because an extra call is now allowed. Restore the correct source and rerun. This small mutation check asks whether the test can detect the particular defect it claims to cover. It does not mean the suite detects every possible defect.

Avoid asserting incidental details unless they form part of your contract. A test can care that an invalid action never reaches a tool without caring whether validation lives in one function or five. Exact trace order is helpful for this tiny example, but a real concurrent harness might promise only a partial order. Write the requirement first, then decide what evidence establishes it.

Learn in your browser · Chapter 06 of 06

You now have a small contract, an executable harness and tests that exercise important boundaries. The next step is to describe the scope precisely and decide what still needs evidence before connecting real systems.

Use three layers of evidence. Unit tests establish local control-flow rules with controlled dependencies. Adapter integration tests establish that request and response formats, timeouts and error mappings match the actual service. Model evaluation measures whether answers and decisions are useful on representative tasks. Passing the first layer does not substitute for the other two. Keep their reports separate so a large green test count cannot conceal an untested model or network boundary.

Scroll sideways to see every column.

Record the source revision, Python version, commands, fixture version, expected results and observed results. Keep a failure case as well as a happy path. If you change the budget policy, update the specification and add a test that would fail under the old policy. Do not quietly adjust expected outputs just to turn a failing suite green; explain why the new behaviour is correct.

This lab is intentionally synchronous, single-run and in-memory. It has no authentication, concurrent budget reservation, durable checkpoint, worker isolation or real model integration. Its short trace is not a compliance product. The useful deliverable is a repeatable local test pattern and a clear next-test list. You can reuse that pattern in a larger harness while replacing its deliberately limited adapters and controls.