Agent Test Harness

Learn how to test an agent route without calling a real model. @b4run/testing intercepts the model's HTTP endpoint with aimock, a local mock that replays pre-written fixtures. Your tools, prompts, capabilities and state all run normally. Only the LLM is replaced, so the tests are deterministic and safe for CI.

Locally, you can opt into live mode to probe model behaviour, with looser assertions that allow for nondeterminism.

Install#Copy link to section: Install

Install the harness and vitest as dev dependencies:

npm install -D @b4run/testing vitest

First test#Copy link to section: First test

Here is the shape you'll use for most agent tests:

import { fileURLToPath } from "node:url"
import { afterAll, it } from "vitest"
import { createAgentHarness, expectFinalMessage, expectToolCalled, script } from "@b4run/testing"
 
const appRoot = fileURLToPath(new URL("..", import.meta.url))
const h = await createAgentHarness({ appRoot, route: "/chat#agent" })
afterAll(async () => {
  await h.close()
})
 
it("filters open items", async () => {
  const run = await h.run({
    input: "Filter open items",
    fixtures: script()
      .user("Filter open items")
      .callsTool("applyFilter", { status: "open" })
      .replies("Found 2 open items."),
  })
  expectToolCalled(run, "applyFilter").withArgs({ status: "open" })
  expectFinalMessage(run).toContain("Found 2")
}, 60_000)

createAgentHarness boots aimock on a random port, runs typegen and resolves your route, all in the same process. The route string is the Agent Protocol key: "/chat#agent" means the agent export from src/app/chat/index.ts. Await h.close() in afterAll to stop aimock and restore env vars.

Fixture script#Copy link to section: Fixture script

script() compiles one or more fixture groups for fresh-thread scenarios. Each .user() starts a group whose turnIndex begins at zero. .callsTool() tells aimock to respond with a tool call, and .replies() tells it to respond with a text message. For a same-thread follow-up, use explicit fixtures with cumulative turn indexes (shown below):

script()
  .user("Summarize my project")      // aimock matches when user message contains this
  .callsTool("readFile", { path: "README.md" })  // model responds: call readFile
  .replies("Here's a summary…")      // model responds after tool result
  .build()                           // returns AimockFixture[]; or pass builder directly to h.run()

aimock matches fixtures by substring on the latest user message plus turnIndex (count of assistant messages already in the thread). You can skip .build(). Pass the builder directly to h.run({ fixtures }) and B4.run unwraps it for you.

Recipes#Copy link to section: Recipes

Assert a tool was called with specific args#Copy link to section: Assert a tool was called with specific args

Use expectToolCalled with withArgs:

expectToolCalled(run, "applyFilter").withArgs({ status: "open" })

withArgs does a partial (subset) match, so extra args are ignored. Chain .times(n) to assert a call count, or .never() to assert the tool wasn't called.

Assert the final message#Copy link to section: Assert the final message

expectFinalMessage supports substring, regex, and full-text checks:

expectFinalMessage(run).toContain("Found 2")
expectFinalMessage(run).toMatch(/found \d+ items/i)
expectFinalMessage(run).toEqual("Found 2 open items.")

Assert streamed tokens arrived#Copy link to section: Assert streamed tokens arrived

This one throws when nothing streamed:

expectStreamedTokens(run)  // throws if zero tokens were streamed

Multi-turn: run the agent twice#Copy link to section: Multi-turn: run the agent twice

The second run continues the same thread, so its fixture expects turnIndex: 1:

const first = await h.run({
  input: "List tasks",
  fixtures: [
    {
      match: { userMessage: "List tasks", turnIndex: 0, hasToolResult: false },
      response: { content: "You have 3 tasks." },
    },
  ],
})
 
const followUp = await h.run({
  input: "Mark first done",
  fixtures: [
    {
      // The same thread already contains the first assistant message.
      match: { userMessage: "Mark first done", turnIndex: 1, hasToolResult: false },
      response: { content: "Done. You have 2 tasks left." },
    },
  ],
})
 
expectFinalMessage(first).toContain("3 tasks")
expectFinalMessage(followUp).toContain("2 tasks left")

Two h.run() calls without a reset are two turns on the same thread, so the follow-up can exercise thread state.

Fresh-thread test isolation#Copy link to section: Fresh-thread test isolation

Call h.reset() to start over on a new thread:

const original = await h.run({
  input: "Remember project alpha",
  fixtures: script().user("Remember project alpha").replies("Remembered."),
})
 
h.reset() // new thread and a clean fixture set; aimock keeps the same port
 
const isolated = await h.run({
  input: "What project did I mention?",
  fixtures: script().user("What project did I mention?").replies("No project yet."),
})
 
expectFinalMessage(original).toContain("Remembered")
expectFinalMessage(isolated).toContain("No project")

Use h.reset() between scenarios or tests that need a fresh thread. Leave it out between turns of one thread.

Tool output offloading#Copy link to section: Tool output offloading

When large tool outputs are offloaded to a stub file, assert the offload marker:

expectOffloaded(run, "generateReport")

Assert tools were called in order#Copy link to section: Assert tools were called in order

Pass the tool names in the order you expect:

expectToolSequence(run, ["lookupAirport", "computeNavlog", "writeFile"])

By default it's a subsequence match, so other tools may appear between them. Pass { strict: true } to require the tools to appear consecutively:

expectToolSequence(run, ["validate", "save"], { strict: true })

Assert no tool returned an error#Copy link to section: Assert no tool returned an error

HITL permission interrupts don't count as errors here:

expectNoToolErrors(run)

To inspect tool results yourself, use run.toolResults. It's a ReadonlyArray<ObservedToolResult> collected from the run's streamed tool_result events. A tool that threw streams its status: "error" ToolMessage, so isError never depends on the final message list. A stream that carried no tool results falls back to the final thread messages via deriveToolResults (also exported).

Each entry has the shape { name: string, status?: "error" | "success", content: unknown, isError: boolean }. Use it to assert on specific error messages or content when expectNoToolErrors is too broad:

import { deriveToolResults } from "@b4run/testing"
 
const failing = run.toolResults.filter((r) => r.isError)
expect(failing).toHaveLength(0)

State assertions#Copy link to section: State assertions

expectState reads the final agent state:

expectState(run).messages.toHaveLength(3)
expectState(run).field("runningSummary").toBeTruthy()
expectState(run).field("todos").toEqual([{ content: "ship it", status: "pending" }])

run.state is the full agent state after the run, the same object a checkpointer would persist.

Supply the middleware context#Copy link to section: Supply the middleware context

createAgentHarness invokes the route's agent directly, so src/middleware.ts never runs and ctx.middleware is undefined inside tools. Pass middlewareContext with the record your middleware would have returned from allow(context). It's either a fixed value or a function evaluated once per run()/resume() that receives the turn's threadId and input:

const h = await createAgentHarness({
  appRoot,
  route: "/assistant#agent",
  middlewareContext: async ({ threadId }) => ({
    session: await openTestSession(threadId),
    ledger: snapshotLedger(),
  }),
})

Every tool the run invokes reads that record as ctx.middleware. The same option is accepted on defineEval so b4 eval can drive routes whose tools depend on middleware.

Constrain the final message#Copy link to section: Constrain the final message

A Hashbrown client sends a JSON Schema as hashbrown.responseSchema on every AG-UI run, and the server binds it on the root model so the final message must match it. The harness drives the route directly, with no AG-UI request, so pass the same schema as responseSchema:

const h = await createAgentHarness({
  appRoot,
  route: "/assistant#agent",
  responseSchema: {
    type: "object",
    properties: { ui: { type: "array", items: { type: "string" } } },
    required: ["ui"],
    additionalProperties: false,
  },
})

Every turn then sends the model the same response_format a production run sends, so recorded fixtures capture what the model does under that constraint. Without it, a recording can show replies production never produces, such as an empty final message. Only an agent route on a provider that supports schema-constrained output (openai, anthropic) can take one; any other route fails the run instead of running unconstrained. defineEval accepts the same option.

Choose how much to run#Copy link to section: Choose how much to run

@b4run/testing has a separate factory for each level:

• createAgentHarness() runs tools, prompts, capabilities, and state directly through B4.run's runtime in the test process, with no application port. Use it by default for fast agent-behavior tests.
• createAgentProtocolInjector() drives Agent Protocol requests through the runtime's fetch handler without binding a port. Use it when the HTTP request, response, or SSE shape is what you're testing.
• createSubprocessApp() starts a real b4 dev child process. Use it for multi-process, restart, and persistence tests.

Each factory has its own option and result types. There's no mode option on createAgentHarness().

CI setup#Copy link to section: CI setup

You don't need any extra setup. aimock runs on a random port and stops when the harness closes. Just guard that fixture files are committed:

- run: npx vitest --run
- run: test -z "$(git status --porcelain -- test/fixtures/)"   # tracked and untracked drift

Your scaffolded app already has a test#Copy link to section: Your scaffolded app already has a test

The default basic template generates test/agent.test.ts, which imports @b4run/testing and replays one scripted greeting against the /hello#agent route. The navlog template (npm create b4-app@latest my-agent -- --template navlog) generates keyless unit tests under server/test/ for its navlog math, POH tables, weather parsers and tools, and replays its agent through scripted fixtures in server/src/app/navlog/evals/navlog-quality.eval.ts. In either app, run the suite right after scaffolding from the app root:

npm test

To test the route itself, add a harness test that scripts the model and lets the tools run, for example (computeNavlog is pure, so this one needs no network):

import { fileURLToPath } from "node:url"
import { afterAll, it } from "vitest"
import { createAgentHarness, expectFinalMessage, expectToolCalled, script } from "@b4run/testing"
 
const appRoot = fileURLToPath(new URL("..", import.meta.url))
const h = await createAgentHarness({ appRoot, route: "/navlog#agent" })
afterAll(async () => {
  await h.close()
})
 
const navlogInput = {
  aircraft: { tailNumber: "N738ZU", cruiseRpm: 2400, usableFuelGal: 50 },
  altitudeFt: 4500,
  departureTimeUtc: "2026-10-06T14:00:00Z",
  waypoints: [
    { id: "KSTP", lat: 44.9346, lon: -93.0603, elevationFt: 705, magneticVariationDeg: 0, kind: "airport" },
    { id: "KRST", lat: 43.9083, lon: -92.49, elevationFt: 1317, magneticVariationDeg: 0, kind: "airport" },
  ],
  winds: [{ dirDegTrue: 320, speedKt: 20 }],
}
 
it("computes the navlog and cites the POH", async () => {
  h.reset()
  const run = await h.run({
    input: "Plan KSTP to KRST at 4500 feet, departing 1400Z.",
    fixtures: script()
      .user("Plan KSTP to KRST at 4500 feet, departing 1400Z.")
      .callsTool("computeNavlog", navlogInput)
      .replies("66 nm, 33 minutes, 5.5 gal. [poh/cruise-performance.md, Figure 5-7]"),
  })
  expectToolCalled(run, "computeNavlog")
  expectFinalMessage(run).toContain("[poh/")
}, 60_000)

Call h.reset() per test. reset() starts a fresh thread and clears the previous scenario's fixtures, so the tests stay isolated.

Grow it by:

1. Adding more script() scenarios: one it block per behaviour you want to pin.
2. Committing fixtures for complex flows: follow Fixtures and Recording for multi-turn or multi-tool scenarios that are tedious to hand-write.

Related#Copy link to section: Related