Testing agent loops: budgets, termination and failure recovery

Read a sample · 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.

Read a sample · 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.

Read a sample · 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.

Read a sample · 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.