inspect-test-utils 1.4.2

Simple models and tasks for integration testing

inspect-test-utils

A small collection of tasks, scorers, and a simple model for use with the Inspect AI framework. It is designed to support integration/acceptance tests, demos, and reproductions by providing:

• Ready-made Tasks that exercise common evaluation patterns (simple generation, numeric closeness, failure injection, and sandbox configuration).
• Scorers for deterministic or parameterized scoring (including hardcoded outputs and a logarithmic closeness score).
• A hardcoded ModelAPI implementation that can deterministically emit tool calls and/or final answers, useful for testing tool-calling flows without hitting external APIs.

Passing arguments Most tasks and the hardcoded model accept parameters. With the Inspect CLI you can pass them via --task-arg and --model-arg repeatedly:

• Example: make the task generate 3 samples and set a numeric target for guessing: inspect eval inspect_test_utils/guess_number
  --task-arg sample_count=3
  --task-arg target=42.7
  --model hardcoded --model-arg answer=42.6

What’s included

• Tasks (inspect_test_utils.tasks)

  • say_hello(sample_count=1): Simple task; expects a response that includes "hello".
  • guess_number(sample_count=1, target="42.7"): Uses a logarithmic closeness scorer for numeric answers.
  • hardcoded_score(sample_count=10, hardcoded_score=None, hardcoded_score_by_sample_id_and_epoch=None): Scores are injected from parameters; useful for testing aggregations and edge cases (including NaN).
  • sometimes_fails_setup(sample_count=10, fail_setup_on_epochs=None, failure_rate=0.2): Randomly raises during setup via a failing solver; useful to test retry/resume behavior.
  • sometimes_fails_scoring(sample_count=10, fail_score_on_epochs=None, failure_rate=0.2): Randomly raises during scoring; useful to test scorer error handling.
  • configurable_sandbox(sample_count=1, cpu=0.5, memory="2G", storage="2G", gpu=None, gpu_model=None, allow_internet=False): A task with runtime configurable sandbox.

• Scorers (inspect_test_utils.scorers)

  • failing_scorer(fail_on_epochs=None, failure_rate=0.2): Raises errors at a controlled rate for selected epochs.
  • closeness_log(): Scores 1.0 for exact equality, otherwise 1/(1+log1p(relative_error)) for numeric strings.
  • hardcoded_scorer(hardcoded_score=None, hardcoded_score_by_sample_id_and_epoch=None): Returns pre-specified Score objects or looks them up by sample id and epoch.

• Model (inspect_test_utils.hardcoded)

  • hardcoded: A ModelAPI that can emit a sequence of tool calls (e.g., bash) for a number of repetitions and then submit a final answer. Parameters include:
    • answer: final answer string (default: "done").
    • repetitions: how many tool-call "turns" before submitting.
    • tool_calls: list of tool calls or shell strings (e.g., ["echo hi", "ls -la"]) to simulate; defaults to none.
    • delay: optional delay (seconds) before returning each model output.

Testing checkpoint/resume of an (agent, task) pair

inspect_test_utils includes a crash/resume harness that verifies an (agent, task) pair correctly checkpoints and resumes after a mid-run or scoring crash. Use it with any task that calls react() (or another checkpointer-aware solver) and has a checkpoint trigger configured.

from inspect_ai import Task
from inspect_ai.agent import react
from inspect_ai.dataset import Sample
from inspect_ai.scorer import includes
from inspect_ai.util import CheckpointSampleConfig

from inspect_test_utils import (
    run_resume_test,
    after_turns,
    at_scoring,
    assert_resumed,
    assert_agent_not_restarted,
    assert_score_recovered,
)

task = Task(
    dataset=[
        Sample(
            id="s1",
            input="go",
            target="done",
            checkpoint=CheckpointSampleConfig(sandbox_paths={"default": ["/root"]}),
        )
    ],
    solver=react(...),
    scorer=includes(),
    sandbox="docker",
)

# Crash mid-run (after the 2nd sandbox exec) and assert the agent resumed:
r = run_resume_test(task, crash=after_turns(2), compute_baseline=False)
assert_resumed(r)

# Crash at the first scoring call and assert the agent was NOT re-run (scoring-only resume):
task_no_sandbox = Task(
    dataset=[Sample(id="s1", input="hi", target="hi")],
    solver=react(...),
    scorer=includes(),
)
r = run_resume_test(task_no_sandbox, crash=at_scoring())
assert_resumed(r)
assert_agent_not_restarted(r)  # agent loop skipped on scoring resume
assert_score_recovered(r)      # score matches baseline

Both after_turns(n) (crash after the n-th sandbox exec) and at_scoring() (crash at the first scorer call) are supported. after_turns requires compute_baseline=False because the exec patch is incompatible with a second in-process checkpointed eval.

To bound an open-ended agent (a real model that won't submit on its own), pass message_limit= / time_limit= — these are forwarded to eval_set as eval-level limits that are recreated per attempt, so they survive the resume. (An as_solver(limits=[message_limit(...)]) does not: the same Limit instance is reused on the resume attempt and raises "a Limit may only be used once".) With a real model also set a request timeout on the model — e.g. get_model(name, config=GenerateConfig(timeout=90)) — so a hung provider call can't wedge the run.

run_resume_test uses an in-process soft crash (CrashInjected exception + eval_set retry), suitable for CI. For a true os._exit crash as in a real k8s/Hawk deployment, use the hard=True injector instead — see below.

Crash + resume on a real deployment (Hawk)

To exercise crash + resume of a real agent on a platform that handles restart (k8s / Hawk), use the registered crashing_react solver — chain(crash_after_exec(n, hard=True), react(...)). On the n-th agent bash call it calls os._exit; the platform relaunches the sample and resumes it from its last checkpoint.

The injector is resume-safe: it arms only on the initial attempt (read from the sample's checkpoint attempt) and disarms itself on resume, so the wrapper can stay in the config the platform replays on resume — a deployment cannot swap solvers without breaking hydration — and the resumed run completes instead of re-crashing.

crashing_react is registered for plugin discovery (inspect_test_utils/crashing_react), so an eval-set can reference it from its solvers: block:

solvers:
  - package: "git+https://github.com/METR/inspect-test-utils"   # a version/tag exporting crashing_react
    name: inspect_test_utils
    items:
      - name: crashing_react
        args:
          crash_after: 8   # os._exit on the 8th agent bash call
          hard: true
checkpoint:
  enabled: true
  trigger: { type: turn, every: 1 }

Launch the eval-set, confirm a checkpoint fired before the crash (e.g. hawk trace <id>), then resume after the crash (hawk eval-set resume <id> --secret …); the resumed sample hydrates from its last checkpoint and runs to completion. See docs/user-guide/checkpointing.md in the Hawk repo for the resume workflow and requirements.

Never run crashing_react(hard=True) (or crash_after_exec(n, hard=True)) inside a pytest process — os._exit would kill the test runner. hard=True is for real eval-set jobs only; for an in-process test use run_resume_test (above) or pass hard=False.

To compose the injector with a different agent or tool set, build the chain yourself: chain(crash_after_exec(n, hard=True), as_solver(react(tools=[...]))). This is the resume_probe pattern generalised to any agent via the shared sandbox-exec seam.

Installation

pip install inspect-test-utils

License

MIT. See LICENSE.