Runtime Testing

Test agent runtime wiring with mock models, stubbed tools, real thread state, and pending actions.

Agent runtime tests run the real PAI runtime with controlled model and tool behavior. Use them when the behavior depends on lifecycle hooks, active-tool filtering, context mapping, thread state, ToolData projection, pending actions, storage, or multi-turn flow.

The normal shape is:

mock the model;
run the real runtime;
use real tools when tool/runtime integration matters;
stub tools when the test only cares about runtime wiring.

import {
  createTestUtils,
  createTestRuntime,
  stubTools,
} from "@pai/test-utils";
import { myAgent, type MyContract } from "./agent";

const t = createTestUtils<MyContract>();

const testIdentity = {
  identity: { id: "test-user", workspaceId: "test-workspace" },
};

The test utils give you typed model builders and state selectors. createTestRuntime is a constructor convenience around createAgentRuntime: it creates the same runtime surface with test defaults such as in-memory storage and an optional model override.

Use the normal runtime client API in tests. That keeps runtime tests close to production usage.

Typed Helpers

createTestUtils<Contract>() binds your agent contract once so runtime test helpers stay typed everywhere else. It is not a runtime and it is not your agent. It is a contract-bound helper surface for building mock model responses and reading public thread state.

Helper	Use
`t.model.queue(...)`	Create a mock model with deterministic queued responses.
`t.model.imperative()`	Create a mock model you drive one call at a time.
`m.response([...])`	Group several model parts into one response.
`m.text(...)` / `m.textStream(...)`	Build text response parts.
`m.toolCall(name, input)`	Build a typed tool-call part.
`t.getTool(state, name)`	Read one typed tool item from thread state.
`t.tools(state, name)`	Read all typed tool items for a tool name.
`t.waitForTool(thread, name, predicate?)`	Wait until a thread contains a matching tool item.
`t.actionName(toolName, suspendName)`	Build a typed suspend action name when a lower-level API needs the string.

Use createTestUtils when the helper depends on your contract. Use ordinary test framework mocks for app services, repositories, or tool stubs.

A failed tool carries its whole failure on the part as errorText. There is no structured second half to cross-check, so most assertions want the part alone:

expect(t.findTool(state, 'search_matters')).toMatchObject({
  state: 'output-error',
  errorText: expect.stringContaining('Tool is disabled'),
})

To assert on a pending action rather than a tool part, use findPending(thread, state, actionName) — it walks the transcript itself, so a test never has to pair a part back to its owning message by hand.

Runtime Wiring

This is the core runtime test: the model is mocked, the tool body is stubbed, and the real runtime turns the model tool call into public thread state.

const model = t.model.queue((m) => [
  m.response([
    m.toolCall("searchAgencies", {
      capability: "observability",
      region: "North America",
    }),
  ]),
  m.response([m.text("I found Northstar.")]),
]);

const agent = stubTools(myAgent, {
  unhandled: "fail",
  tools: {
    searchAgencies: async ({ input }) => ({
      agencies: [
        {
          id: "agency-northstar",
          name: "Northstar Engineering",
          matchedCapability: input.capability,
        },
      ],
    }),
  },
});

await using runtime = createTestRuntime({ agent, model });
const client = runtime.client(testIdentity);
const thread = client.thread(client.newThreadId());
const run = await thread.send("Find observability agencies");
const state = await run.waitUntilIdle();

const tool = t.getTool(state, "searchAgencies");
expect(tool.output?.agencies[0]?.id).toBe("agency-northstar");

This proves the runtime path: model response parsing, tool-call execution, schema validation, public PaiMessage projection, and final thread state. It does not prove that the real model would choose searchAgencies; the model choice is mocked.

Multi-Turn Workflows

Keep the runtime open when the test needs multiple sends or several model responses.

const model = t.model.queue((m) => [
  m.response([
    m.toolCall("getJobBrief", { jobBriefId: "job-telemetry-platform" }),
  ]),
  m.response([m.text("Loaded the brief.")]),
  m.response([
    m.toolCall("compareProposals", { jobBriefId: "job-telemetry-platform" }),
  ]),
  m.response([m.text("Northstar is ranked first.")]),
]);

const agent = stubTools(myAgent, {
  unhandled: "fail",
  tools: {
    getJobBrief: async () => ({ title: "Warehouse telemetry platform" }),
    compareProposals: async () => ({
      rankings: [{ agency: { id: "agency-northstar" }, totalScore: 94 }],
    }),
  },
});

await using runtime = createTestRuntime({ agent, model });
const client = runtime.client(testIdentity);
const thread = client.thread(client.newThreadId());

const loadRun = await thread.send("Load the brief");
await loadRun.waitUntilIdle();

const compareRun = await thread.send("Now compare proposals");
const state = await compareRun.waitUntilIdle();

expect(t.getTool(state, "compareProposals").output?.rankings[0]?.agency.id).toBe(
  "agency-northstar",
);

Use this for lifecycle hooks, commands, queueing, thread history, follow-up turns, and several tools interacting through one thread.

Mock Model Control

Queued responses are best for ordinary runtime tests:

const model = t.model.queue((m) => [
  m.response([
    m.text("I will check that."),
    m.toolCall("lookupShipment", { orderId: "o_42" }),
  ]),
  m.response([m.text("Shipment o_42 is ready.")]),
]);

Use imperative responses when timing or prompt inspection matters:

const model = t.model.imperative();
const agent = stubTools(myAgent, {
  unhandled: "fail",
  tools: {
    lookupShipment: async ({ input }) => ({
      orderId: input.orderId,
      status: "ready",
    }),
  },
});

await using runtime = createTestRuntime({ agent, model });

const client = runtime.client(testIdentity);
const thread = client.thread(client.newThreadId());
const runPromise = thread.send("Find order o_42");
const call = await model.nextCall();

expect(call.prompt.at(-1)?.content).toBeDefined();
call.respond((m) =>
  m.response([m.toolCall("lookupShipment", { orderId: "o_42" })]),
);

await (await runPromise).waitUntilIdle();

Queued and imperative mocks share the same response parts: m.text, m.textStream, m.toolCall, and m.response.

Suspend And Resume

Pending actions are runtime behavior. Test them through a real thread, even when the tool body is stubbed.

const model = t.model.queue((m) => [
  m.response([
    m.toolCall("approveAwardRecommendation", {
      jobBriefId: "job-telemetry-platform",
      approverName: "Priya Shah",
    }),
  ]),
  m.response([m.text("Award approved.")]),
]);

const agent = stubTools(myAgent, {
  unhandled: "fail",
  tools: {
    approveAwardRecommendation: async ({ suspend }) =>
      suspend("approval", {
        message: "Approve this award recommendation?",
      }),
  },
});

await using runtime = createTestRuntime({ agent, model });
const client = runtime.client(testIdentity);
const thread = client.thread(client.newThreadId());
const run = await thread.send("Approve the award.");

const pending = await run.waitForPending({
  toolName: "approveAwardRecommendation",
  suspendName: "approval",
});

await pending.submit({ approved: true, comment: "Approved." });
const state = await run.waitUntilIdle();

expect(t.getTool(state, "approveAwardRecommendation").output?.approved).toBe(
  true,
);

Use this to prove the pending action is projected into thread state and resumes the active run. Use an isolated tool test when you only need to prove the tool returns a suspend payload for a given input.

Context, Scope, And Storage

Pass default identity and client data to createTestRuntime when one setup is enough for the test:

await using runtime = createTestRuntime({
  agent,
  model,
  identity: {
    identity: { id: "alice", role: "admin", workspaceId: "ws-1", region: "NA" },
  },
  clientData: {
    locale: "en-AU",
  },
});

const thread = runtime.newThread();

This keeps the production runtime-context path intact: the runtime calls the configured runtimeContext builder with the resolved identity, threadId, and signal before each execution episode. Use runtime.client(...) when a test needs a different identity for a specific client:

const client = runtime.client({
  identity: { id: "alice", role: "admin", workspaceId: "ws-1", region: "NA" },
  clientData: { locale: "en-AU" },
});

Partitioning

Every client in a test runtime shares one storage partition by default, so two identities can address the same thread without any setup. That is what most tests want, and it is what makes a multi-user conversation testable at all.

A test that asserts the opposite — that one caller cannot reach another's thread — has to declare the partition, because the default deliberately does not isolate:

await using runtime = createTestRuntime({
  agent,
  model,
  // Mirror production: partition per user, so a second caller is a second tenant.
  scopeKey: (identity) => identity.userId,
});

const owner = runtime.client({ identity: { userId: "user-1" } });
const other = runtime.client({ identity: { userId: "user-2" } });
// `other` cannot see `owner`'s threads.

Declare the same derivation the application passes to createPai, or the test proves isolation the product does not actually have.

runtime.testing.scopeKey(identity?) returns the key a given identity resolves to, defaulting to the runtime's own identity — use it when addressing the store directly.

Use overrideRuntimeContext only when the test is intentionally not testing the agent's runtime-context builder:

await using runtime = createTestRuntime({
  agent,
  model,
  identity: {
    identity: { id: "alice", workspaceId: "ws-1" },
  },
  overrideRuntimeContext: {
    workspace: { id: "ws-1", plan: "enterprise" },
    featureFlags: ["award-recommendations"],
  },
});

Pass real or fake storage when the behavior is storage-specific:

await using runtime = createTestRuntime({
  agent,
  model,
  storage: createInMemoryThreadStore(),
});

Capability Bindings

When the agent's tools declare capabilities, pass test bindings through capabilities. The production agent stays verbatim; only the binding record changes:

import { sandboxCapability } from "@pai/sandbox";
import { createInMemorySandboxProvider } from "@pai/sandbox-inmemory";

await using runtime = createTestRuntime({
  agent,
  model,
  capabilities: {
    sandbox: sandboxCapability({ provider: createInMemorySandboxProvider() }),
  },
});

This runs the real capability path — key derivation, per-call acquisition, ctx.capabilities wiring, and dispose-on-thread-delete — against an in-memory implementation. Use runTool with direct facades when only the tool body is under test.

Runtime Testing

On this page