Agent Test Harness

Agent tests in CI should be deterministic and replay committed fixtures. @b4run/testing achieves this by intercepting 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. Optional, explicitly gated local smoke tests may instead use a live model.

For CI, the fixture is the source of truth. Use live mode locally to probe model behaviour, with looser assertions that account for nondeterminism.

Install

pnpm add -D @b4run/testing vitest

First test

This is the full 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

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; .replies() tells it to respond with a text message. For a same-thread follow-up, use explicit fixtures with cumulative turn indexes as 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 do not need to call .build() — pass the builder directly to h.run({ fixtures }) and B4.run unwraps it for you.

Recipes

Assert a tool was called with specific args

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

withArgs does a partial (subset) match — extra args are ignored. Chain .times(n) to assert exact call count, or .never() to assert the tool was not called.

Assert the final message

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

Assert streamed tokens arrived

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

Multi-turn: run the agent twice

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 conversation state.

Fresh-thread test isolation

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 require a fresh thread, not between turns in one conversation.

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

expectToolSequence(run, ["searchCorpus", "readDoc", "writeFile"])

Asserts that the named tools were called in that order (subsequence match by default — other tools may appear between them). Pass { strict: true } to require contiguity: the tools must appear consecutively with nothing else in between.

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

Assert no tool returned an error

expectNoToolErrors(run)

Asserts that no tool returned an error result. HITL permission interrupts are not counted as errors.

To inspect tool results manually, use run.toolResults — a ReadonlyArray<ObservedToolResult> derived from the final conversation messages by deriveToolResults (also exported). Each entry has the shape { name: string, status?: "error" | "success", content: unknown, isError: boolean }. This is useful for asserting 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

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.

Choose the execution boundary

@b4run/testing exposes a separate factory for each execution boundary:

• createAgentHarness() runs tools, prompts, capabilities, and state directly through B4.run's runtime in the test process. It is the default choice for fast agent-behavior tests and does not bind an application port.
• createAgentProtocolInjector() drives Agent Protocol requests through the runtime's fetch handler without binding a port. Use it when the HTTP request, response, or SSE contract is the behavior under test.
• createSubprocessApp() starts a real b4 dev child process. Use it for process-boundary, restart, and persistence tests.

These factories deliberately have separate option and result types. Choose the factory for the boundary the test needs; createAgentHarness() does not accept a mode option.

CI setup

No extra setup needed. @b4run/testing is CI-safe by default: aimock runs on a random port and stops when the harness closes. The only thing to guard is that fixture files are committed:

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

Fixture files: author, commit, replay

Fixtures and Recording is the canonical workflow.

Author inline and snapshot to a file

See Fixtures and Recording.

Record from a real model (local only)

See Fixtures and Recording.

Replay a fixture file in tests

See Fixtures and Recording.

Live mode (real model)

See Fixtures and Recording.

Your scaffolded app already has a test

create-b4-app generates server/test/research.test.ts that imports @b4run/testing and covers the default /research#agent route. Run it immediately after scaffolding, from the workspace root:

npm test

The generated suite starts by verifying the corpus search and citation path:

import { fileURLToPath } from "node:url"
import { afterAll, it } from "vitest"
import type { FixtureSet } from "@b4run/testing"
import {
  createAgentHarness,
  expectFinalMessage,
  expectInterrupt,
  expectOffloaded,
  expectSubagent,
  expectToolCalled,
  script,
} from "@b4run/testing"
 
const appRoot = fileURLToPath(new URL("..", import.meta.url))
const h = await createAgentHarness({ appRoot, route: "/research#agent" })
afterAll(async () => {
  await h.close()
})
 
it("searches the corpus and writes a cited answer", async () => {
  h.reset()
  const run = await h.run({
    input: "What are common agent architectures?",
    fixtures: script()
      .user("What are common agent architectures?")
      .callsTool("searchCorpus", { query: "agent architectures" })
      .callsTool("readDoc", { path: "corpus/agent-architectures.md" })
      .replies("ReAct and plan-and-execute are common. [corpus/agent-architectures.md]"),
  })
  expectToolCalled(run, "searchCorpus")
  expectToolCalled(run, "readDoc")
  expectFinalMessage(run).toContain("[corpus/")
}, 60_000)

The suite also exercises:

• Recall of seeded durable research preferences.
• Candidate writes through remember, plus CLI approval and recall from a fresh thread.
• Researcher subagent dispatch with access to the shared corpus tools.
• Offloading of a large readDoc result.
• The HITL permission gate for runBash, followed by resume.

These scenarios share one harness and call h.reset() per test — reset() starts a fresh thread and clears the previous scenario's fixtures, so the tests stay isolated. The separately gated server/test/sandbox-docker.test.ts verifies that shared tools operate inside an isolated Docker workspace without leaking files to the host or a fresh thread sandbox.

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