Mock Transport

The mock transport is a test and demo transport that implements AgentClientTransport and adds control APIs for seeded state, canned responses, watch events, files, commands, and inspection.

It does not implement a second client facade. It plugs into the real client:

const transport = createMockAgentTransport<MyContract>();

const client = createAgentClient(transport);

From application code's point of view, client is a normal PAI client. The mock only replaces the backend boundary.

app code
  -> real AgentClient
  -> real Thread and RunHandle
  -> mock AgentClientTransport
  -> in-memory store and scripted response queue

Use runtime tests when you need real agent execution, lifecycle hooks, model calls, or tool execution. Use the mock transport when you want a deterministic client-visible backend without importing a server agent.

For convenience, createMockAgentClient() creates both the transport and the real client facade:

const mock = createMockAgentClient<MyContract>();

mock.responses.queue((r) => r.assistant([r.text("Done.")]));

const run = await mock.client.thread("thread-1").send("Hello");
await run.waitUntilIdle();

API Shape

The transport object is both an AgentClientTransport and a controllable test harness:

type MockAgentTransport<
  TContract extends AgentContract,
> = AgentClientTransport<TContract> & {
  seed(seed: MockTransportSeedInput<TContract>): void;
  reset(seed?: MockTransportSeedInput<TContract>): void;

  responses: MockResponseController<TContract>;
  events: MockEventController<TContract>;
  commands: MockCommandController<TContract>;
  files: MockFileController;
  inspect: MockInspectionController<TContract>;
};

The mock client fixture exposes those controls next to the real client:

type MockAgentClient<TContract extends AgentContract> = {
  client: AgentClient<TContract>;
  transport: MockAgentTransport<TContract>;

  seed: MockAgentTransport<TContract>["seed"];
  reset: MockAgentTransport<TContract>["reset"];
  responses: MockAgentTransport<TContract>["responses"];
  events: MockAgentTransport<TContract>["events"];
  commands: MockAgentTransport<TContract>["commands"];
  files: MockAgentTransport<TContract>["files"];
  inspect: MockAgentTransport<TContract>["inspect"];
};

The control APIs are intentionally outside AgentClientTransport. Production transports do not have them; the shared client facade ignores them.

manifest is optional. The mock uses a minimal default manifest unless the code under test calls client.getManifest() and needs realistic client-visible agent metadata:

const mock = createMockAgentClient<MyContract>({
  manifest: myGeneratedContractManifest,
});

The manifest is not the server agent. It should contain only serializable client-safe metadata such as the agent name, tool descriptors, suspend/data channels, and commands.

The mock has no server schema to compute run defaults. When the contract's metadata has required output fields, provide the canonical initial value once for every dynamically created run:

const mock = createMockAgentClient<MyContract>({
  initialMetadata: { rating: null },
});

The option is optional when {} satisfies the client-readable metadata contract. Seeded run fixtures still accept their own typed metadata, so a test can model later values explicitly.

Tool provenance normally comes from the manifest. Use registeredToolMetadata for request-time or test-only registrations that the manifest cannot describe:

const mock = createMockAgentClient<MyContract>({
  registeredToolMetadata: new Map([
    [
      "lookup_order",
      { executor: "server", toolId: "lookupOrder" } as const,
    ],
    ["browser_upload", { executor: "client" } as const],
  ]),
});

A stable server definition produces a typed tool-${name} part with toolMetadata.pai.toolId. A registration without a stable definition id uses the SDK's dynamic-tool part and retains its name and executor provenance. createClientHarness derives this map from a generated React contract, so React tests rarely pass it directly.

Basic Usage

Seed the mock, queue a response, then interact with the client normally:

const mock = createMockAgentClient<EngineeringSourceContract>();

mock.seed((s) => {
  const seeded = s.run({
    threadId: "thread-telemetry",
    messages: [
      s.input("Find observability agencies in North America."),
      s.output([
        s.text("I found Northstar Engineering and Orbit Labs."),
      ]),
    ],
  });

  return {
    threads: [
      s.thread("thread-telemetry", {
        title: "Warehouse telemetry platform",
        messages: seeded.messages,
        runs: [seeded.run],
      }),
    ],
  };
});

mock.responses.queue((r) =>
  r.assistant([
    r.text("Northstar is ranked first for observability work."),
    r.tool("compareProposals").result({
      input: { jobBriefId: "job-telemetry-platform" },
      output: {
        rankings: [{ agencyId: "agency-northstar", totalScore: 94 }],
      },
    }),
  ]),
);

const thread = mock.client.thread("thread-telemetry");
const run = await thread.send("Compare the proposals.");
const state = await run.waitUntilIdle();

state.messages; // PaiMessage<EngineeringSourceContract>[]
state.runs; // normalized run lifecycle

thread.send(...) owns the user input. It appends the user message, creates a run, marks the thread running, consumes the next queued mock response, emits assistant output, and settles the run. Do not include the user message again in the queued response.

Seeded State

seed() creates exact native transport state. The callback builder makes the common case concise while keeping messages and runs as separate normalized collections:

mock.seed((s) => {
  const seeded = s.run({
    threadId: "thread-1",
    runId: "run-award",
    status: "waiting",
    messages: [
      s.input("Draft the recommendation."),
      s.output([
        s.text("I need approval before finalizing."),
        s.tool("approveAwardRecommendation").waiting({
          input: {
            jobBriefId: "job-1",
            recommendedAgencyId: "agency-1",
          },
          action: {
            name: "approval",
            input: { message: "Approve the award recommendation?" },
            id: "action-1",
          },
        }),
      ]),
    ],
  });

  return {
    threads: [
      s.thread("thread-1", {
        title: "Award memo",
        metadata: { projectId: "project-1" },
        messages: seeded.messages,
        runs: [seeded.run],
      }),
    ],
    files: [
      s.file("file-brief", {
        filename: "brief.pdf",
        mediaType: "application/pdf",
        body: new Blob(["brief contents"]),
      }),
    ],
  };
});

s.input(), s.context(), and s.output() build native wire messages with the correct producer and relation metadata. s.run() assigns one run identity to the ordered messages and returns { run, messages }; pass both collections to s.thread(). A run fixture requires at least one input message.

s.assistant([...]) builds a richer model message directly, while s.standalone(message) removes run ownership for intentionally administrative history. There is no cross-part identifier: native tools use toolCallId, data parts use id, and text parts are ordered within their owning message.

Because messages is ordered, fixtures can represent lifecycle context exactly where it entered a run:

s.run({
  threadId: "thread-1",
  messages: [
    s.input("Draft the recommendation."),
    s.output([s.text("I am checking the award constraints.")]),
    s.context("The approval threshold is now $50,000.", { role: "system" }),
    s.output([s.text("This recommendation requires approval.")]),
  ],
});

Use s.input(text, { producer: "trigger" }) together with initiator: "trigger" for trigger-created runs. The default initiator is "send". Use s.output(parts, { producer: "lifecycle" }) for lifecycle-authored output:

s.run({
  threadId: "thread-1",
  initiator: "trigger",
  messages: [
    s.input("Refresh the recommendation", { producer: "trigger" }),
    s.output([s.text("Visible result")], { producer: "lifecycle" }),
  ],
});

Waiting helpers emit an input-available native tool part plus a reserved suspension data part. s.thread() derives the transport's action control state from those chunks. Through the real client, the reserved chunk is projected onto part.suspensions and part.pendingSuspension; the command remains a sidecar resolved from the exact owning pair:

import { isPaiToolPart } from "@pai/protocol";

const thread = mock.client.thread("thread-1");
const state = await thread.refresh();
const message = state.messages.find((item) => item.role === "assistant");
const part = message?.parts.find(isPaiToolPart);

if (!message || !part) throw new Error("Expected a waiting tool");

const action = thread.getPendingAction(message, part);
if (!action) throw new Error("Expected a pending action");

mock.responses.queue((r) => r.assistant([r.text("Approval recorded.")]));
const resumed = await action.submit({ approved: true });
await resumed.run?.waitUntilIdle();

mock.inspect.thread(...) intentionally returns the raw transport snapshot, including reserved data-pai-* chunks. thread.refresh() and thread.getState() return the projected public PaiMessage[] view.

reset(seed?) clears store state, response queues, live watchers, request logs, and generated ids. Passing a seed immediately loads the next fixture.

Response Queue

Only generation actions consume response scripts:

• thread.send(...);
• thread.retry(...);
• thread.regenerate(...).

Ordinary state operations such as renameThread, deleteThread, uploadFile, readFile, executeCommand, and writeToolData mutate the local store directly and do not consume the response queue.

Queue assistant responses for successful generations:

mock.responses.queue((r) =>
  r.assistant([
    r.text("Northstar is ranked first for observability work."),
    r.tool("compareProposals").result({
      input: { jobBriefId: "job-telemetry-platform" },
      output: {
        rankings: [{ agencyId: "agency-northstar", totalScore: 94 }],
      },
    }),
  ]),
);

The response builder is a concise scripting DSL, not another stored message format. The mock converts it to native text and tool parts before publishing the assistant PaiMessage. Use mock.responses.size() to inspect the number of unclaimed scripts and mock.responses.clear() to discard them.

Waiting tool fixtures separate model-provided tool input from the pending action input the client submits. The tool name is already bound by r.tool("approveAwardRecommendation"), so action.name is the suspend/action name and the mock transport derives the full pending action name.

mock.responses.queue((r) =>
  r.assistant([
    r.text("The award memo is ready for approval."),
    r.tool("approveAwardRecommendation").waiting({
      input: {
        jobBriefId: "job-telemetry-platform",
        recommendedAgencyId: "agency-northstar",
      },
      action: {
        name: "approval",
        input: { message: "Approve the award recommendation?" },
      },
    }),
  ]),
);

Client-executed tools (hasExecute: false) wait on an implicit action named output. Its action input has the same shape as the tool input, and the mock omits the raw suspendName discriminator just like the runtime:

mock.responses.queue((r) =>
  r.assistant([
    r.tool("browserUpload").waiting({
      input: { fileName: "receipt.pdf" },
      action: {
        name: "output",
        input: { fileName: "receipt.pdf" },
      },
    }),
  ]),
);

If a backend-executed tool instead declares a suspend schema literally named output, add declaredSuspend: true. This preserves output as the raw suspend discriminator and types action.input from that suspend schema. The same distinction applies to queued, seeded, and streamed waiting fixtures.

await response.tool("reviewJob", {
  input: { jobId: "job-1" },
  waiting: {
    action: {
      name: "output",
      declaredSuspend: true,
      input: { prompt: "Approve this job?" },
    },
  },
});

Queue run errors for accepted generations that should fail:

mock.responses.queue((r) => r.error(new Error("Provider failed")));

const run = await thread.send("Draft the memo.");
const state = await run.waitUntilIdle();

expect(state.error).toMatchObject({
  message: "Provider failed",
  runId: run.runId,
  operation: "send",
  messageId: expect.any(String),
});

A queued run error is consumed by the next admitted send, retry, or regenerate run. The mock appends the user message, creates the run, consumes the queued error, marks the run failed, clears the active run, sets the public ThreadState.error, emits generation.failed, and leaves enough error context for thread.retry() to regenerate the failed user turn.

Run errors are different from tool errors. A run error fails the generation and sets ThreadState.error:

mock.responses.queue((r) => r.error(new Error("Provider failed")));

A tool error is part of an assistant message. It creates a native output-error tool part with errorText and does not, by itself, fail the run:

mock.responses.queue((r) =>
  r.assistant([
    r.tool("searchAgencies").error(new Error("Search service failed")),
  ]),
);

If a generation action runs without a queued response, the run stays running until a response is queued. This makes timing tests straightforward:

const run = await thread.send("Draft the memo.");

expect(thread.getState().thread.status).toBe("running");

mock.responses.queue((r) => r.assistant([r.text("Draft complete.")]));

await run.waitUntilIdle();

Use mock.events only when the behavior is not naturally modeled as a queued response, such as an external thread-list update, a low-level watch event, or a deliberate protocol edge case.

Streamed Responses

Use mock.responses.stream() when the test needs intermediate generation states, not just the final response. The stream controller is consumed by the next send, retry, or regenerate run. If a run is already waiting for a mock response, it attaches immediately.

const response = mock.responses.stream();
const thread = mock.client.thread("thread-1");
const run = await thread.send("Compare proposals.");

expect(await response.waitUntilStarted()).toEqual({
  threadId: "thread-1",
  runId: run.runId,
});

const text = response.text();
await text.delta("Northstar");
await text.delta(" is ranked first.");
await text.end();

await response.complete();
await run.waitUntilIdle();

response.text() and response.tool(name) return imperative controllers. waitUntilStarted() resolves when the response is attached to its concrete thread and run. Callback forms add lifecycle checks around the same controls:

await response.text(async (text) => {
  await text.delta("Northstar");
  await text.delta(" is ranked first.");
  await text.end();
});

Text callbacks must call end(). Tool callbacks must call output(...), waiting(...), error(...), or cancelled(...). Starting another part inside an active scoped callback throws, which keeps tests aligned with the ordered model stream.

Tool input streaming models raw JSON argument chunks from a model provider:

const response = mock.responses.stream();
const run = await mock.client.thread("thread-1").send("Approve the award.");

await response.tool("approveAwardRecommendation", async (t) => {
  await t.input.delta('{"jobBriefId":"job-telemetry-platform",');
  await t.input.delta('"recommendedAgencyId":"agency-northstar"}');
  await t.input.available({
    jobBriefId: "job-telemetry-platform",
    recommendedAgencyId: "agency-northstar",
  });

  await t.running();
  await t.data("progress", { percent: 50 }, { id: "current" });
  await t.output({ approved: true, memoId: "memo-1" });
});

await response.complete();
await run.waitUntilIdle();

t.running() validates that complete input is available. It deliberately does not write a custom running state or emit a chunk because native SDK tool parts remain input-available until output or error arrives. t.cancelled() likewise emits the native tool-output-error transition; cancellation is not stored as a custom tool-part state.

For parts where intermediate states do not matter, use shorthands:

await response.text("I will compare the proposals.");

await response.tool("compareProposals", {
  input: { jobBriefId: "job-telemetry-platform" },
  output: {
    rankings: [{ agencyId: "agency-northstar", totalScore: 94 }],
  },
});

Waiting tools expose pending actions through the real client facade:

import { isPaiToolPart } from "@pai/protocol";

await response.tool("approveAwardRecommendation", {
  input: { jobBriefId: "job-telemetry-platform" },
  waiting: {
    action: {
      name: "approval",
      input: { message: "Approve Northstar for the pilot?" },
    },
  },
});
await response.complete();

const waiting = await run.waitUntilBlocked();
const message = waiting.messages.find((item) => item.role === "assistant");
const part = message?.parts.find(isPaiToolPart);

if (!message || !part) throw new Error("Expected a waiting tool");

const action = thread.getPendingAction(message, part);
if (!action) throw new Error("Expected a pending action");

mock.responses.queue((r) => r.assistant([r.text("Approval recorded.")]));
const resumed = await action.submit({ approved: true });
await resumed.run?.waitUntilIdle();

For tests that do not care about exact JSON chunks but still need input streaming, use the object helper inside a scoped tool callback:

await response.tool("createJobBrief", async (t) => {
  await t.input.stream(largeInput, {
    chunkSize: 80,
    delayMs: 10,
  });
  await t.output({ jobBriefId: "job-telemetry-platform" });
});

The stream publishes the same native chunk grammar consumed by the real client:

Each chunk is carried by a sequenced generation.message.chunk transport event. The mock reduces it into the same native wire message that a real transport would publish, then the client projects public PaiMessage parts. Rendered React hooks, thread.refresh(), thread.getState(), and raw mock.inspect.thread(...) therefore observe the same generation at their respective public and transport boundaries.

response.complete() is always explicit. A tool output does not complete the assistant response by itself, because a real assistant can continue with more text or additional tool calls after a tool result. Use response.fail(error) for a generation failure and response.abort(reason?) to cancel the attached run.

Imperative Events

Most tests should use the response queue. events is the escape hatch for transport-level behavior: reconnection, external state changes, thread-list updates, or watch repair.

const watcher = thread.watch();

mock.events.pushThread("thread-1", {
  type: "state.changed",
});

mock.events.pushThreadHead({
  type: "thread.head.changed",
  reason: "attributes",
  head: renamedThreadHead,
});

These events should flow through watchThread() and watchThreadHeads(). They should not mutate React state directly or bypass the shared client reducer. renamedThreadHead is an exact client-safe ThreadHeadDTO; tests that only need an invalidation can use state.changed and let the client refresh.

events.completeRun(threadId, runId?) and events.failRun(threadId, error) are active-run controls. They operate on the currently active run and throw a clear test-author error if no run is active. Prefer queued responses for normal success and failure fixtures:

mock.responses.queue((r) => r.error(new Error("Provider failed")));

Commands And Files

Commands are client-visible functions, not generation responses. They should be registered as handlers:

mock.commands.set("refreshJobBrief", async ({ input }) => {
  return {
    jobBriefId: input.jobBriefId,
    status: "evaluating",
  };
});

await mock.client
  .thread("thread-1")
  .commands.refreshJobBrief({ jobBriefId: "job-1" });

Files should live in the mock's in-memory file store:

mock.files.add({
  fileId: "file-proposal",
  filename: "proposal.pdf",
  mediaType: "application/pdf",
  body: new Blob(["proposal contents"]),
});

const uploaded = await mock.client.files.upload({
  filename: "notes.txt",
  mediaType: "text/plain",
  body: new Blob(["notes"]),
});

The mock should generate file ids for uploads, preserve metadata, implement readFile(), and return either mock URLs or null from createFileUrl() according to its configuration.

Inspection

Assertions should inspect backend state through a dedicated inspection API, rather than reaching into private maps:

expect(mock.inspect.requests.send()).toHaveLength(1);
expect(mock.inspect.thread("thread-1")?.thread.status).toBe("idle");
expect(mock.inspect.thread("thread-1")?.messages).toEqual(
  expect.arrayContaining([expect.objectContaining({ role: "assistant" })]),
);
expect(mock.inspect.files()).toContainEqual(
  expect.objectContaining({ fileId: "file-proposal" }),
);

Inspection should be read-only. Mutations should go through seed, reset, responses, events, commands, files, or normal client operations.

What The Mock Should Not Do

The mock transport should not:

• call a model;
• execute server tools;
• run lifecycle hooks;
• enforce runtime locks or leases;
• test HTTP, SSE, WebSocket, or IPC framing;
• reimplement AgentClient, Thread, or RunHandle.

Those belong to runtime tests, adapter tests, or transport-specific tests. The mock transport's job is to provide a deterministic backend boundary for the real client facade.

Conformance

The mock transport should pass the same transport conformance suite as other transports, with profiles for watch streams, thread heads, files, commands, and regeneration. That suite should catch drift between the mock and direct/runtime or HTTP-backed clients.

If React testing utilities are built on the mock transport, they should compose like this:

const Pai = createPaiReact<SupportPai>();
const SupportAI = Pai.agent("support");
const t = createReactTestUtils({ Pai, AI: SupportAI });
await using h = t.createClientHarness(options);

render(<App />, { wrapper: h.Provider });

React utilities can wrap this setup, but the underlying behavior should still be the real shared client facade over a conforming mock transport.