Failure Injection Tests for a Tool-Calling Workflow

Verified: capstone/fde-product-discovery-copilot/lessons/M4-L6/m4_l6_fakes.py

def with_retries(
    run: Callable[[LookupArgs, ToolContext], dict[str, Any]],
    attempts: int = 3,
    delays: tuple[float, ...] = (0.1, 0.2),
    retryable: tuple[type[Exception], ...] = (TimeoutError,),
    sleep: Callable[[float], None] = lambda s: None,
) -> Callable[[LookupArgs, ToolContext], dict[str, Any]]:
    """Retry only errors that are worth retrying, a fixed number of times, then let the last error out."""

    def wrapped(args: LookupArgs, ctx: ToolContext) -> dict[str, Any]:
        for attempt in range(1, attempts + 1):
            try:
                return run(args, ctx)
            except retryable:
                if attempt == attempts:
                    raise
                sleep(delays[attempt - 1])
        raise AssertionError("unreachable")

    return wrapped

Three rules sit in those lines:

• Retry only errors that can pass on a second try.
• Cap the attempts.
• Wait between them. Production code usually adds exponential growth and jitter.

The Amazon Builders' Library article on timeouts, retries and backoff with jitter warns that when failures come from overload, retries that add load can make matters significantly worse. It also says APIs with side effects are not safe to retry unless they are idempotent.

Figure 3. A retry inside the tool, then a repeat blocked and a run aborted

How to read this: retries happen inside one step, the duplicate check blocks the same call, and only the step cap stops changing arguments.

Do not trust the tool's output either

A tool can also return something the loop cannot read. The runner summarises a products list with len(...), so a tool that returns {"products": None} raises TypeError outside any try block and the run crashes. The test below is an expected failure: it documents a gap in app/ that this course does not edit.

Verified: capstone/fde-product-discovery-copilot/lessons/M4-L6/test_m4_l6_failure_injection.py

@pytest.mark.xfail(strict=True, raises=TypeError, reason="capstone gap: a malformed tool RESULT is not caught (see the lesson); remove this mark when app/ is fixed")
def test_a_malformed_tool_result_should_not_crash_the_runner(ctx: ToolContext) -> None:
    tool = FakeTool([{"products": None}])  # valid shape for the registry, but the runner calls len() on it
    run(ScriptedProvider([call({"sku": "LUM-VS-100"}), FINAL]), tool, ctx)

With strict=True, the test fails the day someone fixes the runner, which is your cue to remove the mark. Until then, checked_result validates a tool's output inside the tool, so a bad shape becomes an ordinary error observation.

Worked example

Assumptions: illustrative; three attempts, delays of 0.1 s and 0.2 s, each timed-out attempt takes 2 seconds, workflow limit timeout_s=5.0.

One step takes 6.3 seconds, above the 5 second limit. The limit is checked only between steps, so the run ends as timeout after it. With a 5 second limit, cut the attempt timeout to 1 second or allow two attempts.

In the interview: "How do you test an agent?" A good answer has three layers: fakes for the loop's logic (limits, failures, retries), evaluation sets for answer quality, and live provider checks. The first runs on every commit with no model.

Build it

Run the whole folder. No key is needed.

cd capstone/fde-product-discovery-copilot
./.venv/Scripts/python -m pytest lessons/M4-L6 -q

In the wall-clock test, change cost_s to 2.0 and predict the number of tool calls first.

Verified: capstone/fde-product-discovery-copilot/lessons/M4-L6/test_m4_l6_failure_injection.py

def test_the_wall_clock_limit_stops_the_run_between_steps(ctx: ToolContext) -> None:
    clock = FakeClock()
    tool = FakeTool([OK], clock=clock, cost_s=4.0)  # every call "takes" 4 seconds of fake time
    skus = ["LUM-VS-100", "LUM-VS-200", "LUM-VS-300", "LUM-VS-400"]
    result = run(ScriptedProvider([call({"sku": s}) for s in skus]), tool, ctx, timeout_s=5.0, clock=clock)
    assert result.status == "timeout" and tool.calls == 2  # 0 s, then 4 s: both under 5; 8 s: stop
    assert [s.elapsed_ms for s in result.steps] == [4000.0, 4000.0]

With cost_s=2.0 the fake clock reads 0, 2, 4 and 6 at the four step checks. Six is above 5, so the fourth check stops the run: three tool calls, status timeout.

Exercise

A tool returns {"found": {1, 2}}, a Python set, which cannot become JSON. Reproduce the failure and contain it.

Acceptance criteria:

• An expected-failure test with a fake tool that returns a set shows the crash today.
• A wrapper called json_safe makes the same tool end as a failed step with the run completed.
• The wrapper has its own test, and no test sleeps.

Done when both tests pass and the first one is still marked xfail(strict=True). A solution is in lessons/M4-L6/m4_l6_solution_json_safe.py.

Common mistakes

• Testing against the real model or service. Use fakes and keep live checks separate.
• Asserting only the status. Also assert the call counters and the step log.
• Retrying at several layers. Retries in the tool, the client and the loop multiply load.
• Retrying calls that change something. A retry can do the action twice.

Knowledge check

You can now

• Drive WorkflowRunner with fakes.
• Test the five failure classes and assert on counters, status and the step log.
• Put a bounded retry inside a tool and check it against the time limit.
• Explain why a tool's output needs validating too.

Further reading

• How to parametrize fixtures and test functions, pytest documentation: one test, many inputs.
• Timeouts, retries and backoff with jitter, Amazon Builders' Library: limits and idempotency for retries.
• Building effective agents, Anthropic: testing in sandboxed environments.

Last verified: 2026-11-02

Next

This is the last core lesson of the module on bounded agent workflows. The next module starts with Where the Python service sits in the estate.