Testing

Mock mode, `MockToolExecutor`, deterministic seeds, and patterns for unit-testing agents that use `@pleach/core`.

@pleach/core is designed to be tested without a real database or LLM provider account. The combination of mock mode, the in-memory storage/checkpointer pair, and MockToolExecutor covers most agent-level tests; deterministic fingerprints and the audit ledger handle integration tests.

import { MockToolExecutor } from "@pleach/core";
import { MemoryAdapter } from "@pleach/core/sessions";
import { MemorySaver } from "@pleach/core/checkpointing";

Mock mode (one-line setup)

Set the env var; the runtime wires in-memory adapters and a mock executor automatically.

HARNESS_MOCK_MODE=true

const runtime = new SessionRuntime({
  userId: "test-user",
  // No storage / checkpointer / provider — mock mode fills them in.
});

What mock mode wires:

Slot	Wired to
Storage	`MemoryAdapter`
Checkpointer	`MemorySaver`
Provider	A synthesized provider that returns plausible mock text
Tool executor	`MockToolExecutor` (returns synthetic results)

Use it during local dev, in CI for tests that don't need to hit real providers, and as the substrate for example apps.

`MemoryAdapter` + `MemorySaver` for tests

When you want explicit construction (no env vars):

import { SessionRuntime } from "@pleach/core";
import { MemoryAdapter } from "@pleach/core/sessions";
import { MemorySaver } from "@pleach/core/checkpointing";

const runtime = new SessionRuntime({
  storage:      new MemoryAdapter(),
  checkpointer: new MemorySaver(),
  userId:       "test-user",
});

Both are deterministic — same sequence of writes yields the same read state. Tests can await runtime.createSession(), drive a turn, and assert against state without race conditions.

`MockToolExecutor`

Generates plausible mock results based on the tool's output schema. Three modes: synthetic (default), recording (capture real responses for replay), and deterministic (seeded RNG).

import { MockToolExecutor } from "@pleach/core";

const executor = new MockToolExecutor({
  latencyMs: { min: 50, max: 200 },
  deterministicSeed: 12345,
  mockResponses: new Map([
    ["search_corpus", {
      result: { results: [{ id: "doc-abc123", title: "Stub", year: 2024 }] },
    }],
  ]),
});

const result = await executor.execute({
  id: "tc_1",
  name: "search_corpus",
  arguments: { query: "stub query" },
  status: "pending",
});

Options

Field	Type	Purpose
`latencyMs`	`{ min, max }`	Simulated delay range — exercises streaming UI
`recordMode`	`boolean`	Save real responses to disk for later replay
`deterministicSeed`	`number`	Seed for reproducible mock generation
`mockResponses`	`Map<string, MockResponse>`	Explicit per-tool responses
`registry`	`ToolRegistryWrapper`	Tool registry for output schema lookup
`onExecuteStart` / `onExecuteComplete`	callbacks	Lifecycle hooks for assertions

Deterministic mode

Pass a deterministicSeed and the mock executor produces byte-identical results across runs — what you want for snapshot tests:

const executor = new MockToolExecutor({ deterministicSeed: 42 });
const a = await executor.execute(toolCall);
const b = await executor.execute(toolCall);
expect(a).toEqual(b); // identical

The seed feeds an internal RNG that drives the synthetic data generator, so concrete strings like ids, titles, and timestamps stay stable.

Record-and-replay

recordMode: true captures real responses to a JSONL file. Use once against a real provider, then commit the file and switch back to synthetic mode for CI:

const executor = new MockToolExecutor({
  recordMode: true,
  recordPath: "./test/fixtures/tool-responses.jsonl",
});

Pair with MockResponse.fromRecording() to load on the next run:

const responses = MockResponse.fromRecording("./test/fixtures/tool-responses.jsonl");
const executor  = new MockToolExecutor({ mockResponses: responses });

Headless turn driver (`@pleach/core/testing`)

MockToolExecutor mocks the tool layer of a runtime you assemble yourself. The @pleach/core/testing subpath is the other half: it drives a real SessionRuntime turn with no network, no browser, and no real provider, and hands you a structured result. Script the model's output, drive one turn, inspect final text, tool calls, the raw event log, and wall-clock timing. This is the surface for unit tests, regression locks, and benchmark harnesses.

import { runScriptedTurn, scriptText } from "@pleach/core/testing";

// One call: builds a bare runtime, injects a scripted adapter, drives a turn.
const result = await runScriptedTurn({
  prompt: "hello",
  script: scriptText("hello world"),
});

result.text;       // "hello world"  — a real synthesized turn, not a stub
result.ttfbMs;     // time-to-first-delta in ms (benchmark signal)
result.totalMs;    // total drain time in ms
result.eventCount; // number of StreamEvents observed
result.toolCalls;  // tool calls seen this turn
result.events;     // the raw StreamEvent[] for exact-sequence assertions

A multi-element scriptText exercises the streaming/delta path — the concatenation is the final text:

const r = await runScriptedTurn({
  prompt: "hi",
  script: scriptText(["hello", " ", "world"]), // 3 text chunks + a done chunk
});
r.text; // "hello world"

Driving a runtime you already built

runScriptedTurn builds the runtime for you. When you need a runtime with your own plugins, storage, or config, install the scripted adapter with createScriptedAdapter and collect the turn with collectTurn:

import { createScriptedAdapter, scriptText, collectTurn } from "@pleach/core/testing";

runtime.adapter.set(createScriptedAdapter(scriptText("hello world")));
const session = await runtime.sessions.create({
  provider: { type: "anthropic" },
  model: { id: "scripted-model" },
});
const result = await collectTurn(runtime.executeMessage(session.id, "hello"));

Capturing probe emits

Core's default [UXParity:*] probe emit is silent. captureProbes installs a formatter that records the emits a turn produces, so you can assert on them:

import { captureProbes, runScriptedTurn, scriptText } from "@pleach/core/testing";

const cap = captureProbes();
await runScriptedTurn({ prompt: "hello", script: scriptText("hello world") });
const labels = cap.emits.map((e) => e.label);
cap.stop(); // restore the prior formatter

Which helper when

Reach for…	When you want to…
`runScriptedTurn`	the one-liner — a fully synthesized turn from a fixed script, no runtime setup
`createScriptedAdapter` + `collectTurn`	drive a runtime you built (custom plugins, storage, config) with a scripted model
`runProviderTurn` / `createProviderAdapter`	the same, but against a real `AgentProvider` through the production seam
`benchmarkProviders`	sweep N prompts × M providers and compare latency / output size
`captureProbes`	assert on the `[UXParity:*]` probe emits a turn produces
`scriptText` / `textChunk` / `doneChunk`	hand-build the chunk stream (multi-delta, custom `finishReason`, tool calls)

Pinning a turn's behavior as a regression lock

result.events is the raw StreamEvent[], so you can assert on the exact runtime shape — final text, tool sequencing, and the node/channel firing granularity:

import { runScriptedTurn, scriptText } from "@pleach/core/testing";

const r = await runScriptedTurn({ prompt: "hi", script: scriptText("hello world") });

// 1. user-visible output
expect(r.text).toBe("hello world");

// 2. the LLM node actually fired this turn
const llmFired = r.events.some(
  (e) => e.type === "node.fired" && e.node === "llm",
);
expect(llmFired).toBe(true);

// 3. the messages channel took a writer bump (the assistant reply landed)
const wroteMessages = r.events.some(
  (e) => e.type === "channel.write" && e.channel === "messages",
);
expect(wroteMessages).toBe(true);

Run them serially. runScriptedTurn / runProviderTurn / benchmarkProviders swap a process-global module-loader slot per turn (saved and restored around each call), so they must run serially within a process — don't Promise.all them. Back-to-back inside a single test is fine.

Benchmarking real providers

The scripted driver runs a fixed script. To drive a real AgentProvider through the same production seam codepath, wrap it in an adapter with createProviderAdapter and drive a turn with runProviderTurn. The content in result.text comes from provider.execute(...), not a script:

import { runProviderTurn } from "@pleach/core/testing";
import { AnthropicSdkProvider } from "@pleach/core/providers";

const provider = new AnthropicSdkProvider({ apiKey: process.env.ANTHROPIC_API_KEY! });

const result = await runProviderTurn({
  prompt: "Reply with a short greeting.",
  provider,
  model: "claude-sonnet-4-5",
});

result.text;    // the real model's reply
result.ttfbMs;  // time-to-first-delta in ms
result.totalMs; // total drain time in ms

For a runtime you already built, createProviderAdapter implements the same AgentAdapter contract as createScriptedAdapter:

import { createProviderAdapter } from "@pleach/core/testing";

runtime.adapter.set(createProviderAdapter(provider, { model: "claude-sonnet-4-5" }));

benchmarkProviders sweeps several prompts across several providers/models. It runs every prompt against every subject serially and aggregates per-subject latency, output length, and error count:

import { benchmarkProviders } from "@pleach/core/testing";

const report = await benchmarkProviders({
  prompts: ["Reply with a short greeting.", "Count to three."],
  subjects: [
    { label: "anthropic", provider: anthropicProvider, model: "claude-sonnet-4-5" },
    { label: "ai-sdk",    provider: aiSdkProvider,     model: "gpt-4o-mini" },
  ],
});

for (const s of report.subjects) {
  console.log(s.label, s.meanTtfbMs, s.meanTotalMs, s.meanTextLen, s.errorCount);
}

Honest scope. benchmarkProviders measures the raw provider/model dimension — latency (ttfb + total) and output size — over a single non-tool turn. It is not a full domain pipeline: no tool loop, no enrichment, no scoring. For graded offline evaluation (rubrics, datasets, scoring) reach for @pleach/eval. One gotcha: a degenerate or trivial prompt can make the bare graph short-circuit before the LLM call, so send a real prompt.

Asserting against the audit ledger

For tests that verify which model fired, which fallback path ran, or which call class was selected: assert against the AuditableCall rows the runtime emits.

import { MemoryProviderDecisionLedger } from "@pleach/core/audit";

const ledger = new MemoryProviderDecisionLedger();
const runtime = new SessionRuntime({
  storage:                  new MemoryAdapter(),
  userId:                   "test-user",
});

// drive a turn ...

const rows = await ledger.getSession(sessionId);
expect(rows.filter((r) => r.call.callClass === "synthesize")).toHaveLength(1);
expect(rows.filter((r) => r.familyLock !== undefined)).toHaveLength(1);

The exactly-one-synthesize invariant is the easiest test to write and the highest-signal failure when something has drifted.

Asserting against the stream

executeMessage is an async generator — collect events and assert on shapes:

async function collect<T>(iter: AsyncIterable<T>): Promise<T[]> {
  const out: T[] = [];
  for await (const v of iter) out.push(v);
  return out;
}

const events = await collect(runtime.executeMessage(sessionId, "Hello"));

const messageDeltas = events.filter((e) => e.type === "message.delta");
const toolCompleted = events.filter((e) => e.type === "tool.completed");

expect(toolCompleted).toHaveLength(2);
expect(messageDeltas.length).toBeGreaterThan(0);

For streaming tests, prefer assertions on event counts and ordering over assertions on exact deltas — model responses shift across versions even with deterministic seeds.

Fingerprint-based golden tests

Two runs of the same turn with the same input + same package version produce the same fingerprint on the ledger row. That's the snapshot key for replay tests:

import { computeFingerprint } from "@pleach/core";

const fp1 = computeFingerprint(turnInput);
// run turn ...
const fp2 = (await ledger.getSession(sessionId))[0].cacheBreakpoint?.fingerprintComposite;

expect(fp2).toEqual(fp1);

If the fingerprint drifts between runs, something non-deterministic slipped in — a runtime-aware prompt contribution that should have been static, an async stream observer (illegal), or a wall-clock read in a reducer.

Testing plugins

Plugins are pure objects implementing HarnessPlugin — test them in isolation:

const myPlugin = makePlugin();
const contribs = myPlugin.contributePrompts?.() ?? [];

expect(contribs.map((c) => c.id)).toEqual([
  "my-plugin.domain-hint",
  "my-plugin.safety-frame",
]);

For end-to-end plugin tests, register the plugin against a mock-mode runtime and assert on stream events + ledger rows.

CI patterns

Three things keep an agent's tests stable in CI:

HARNESS_MOCK_MODE=true — no provider creds, no real DB.
deterministicSeed on every MockToolExecutor.
Snapshot the ledger, not the stream — token-level deltas shift with provider versions; ledger row shape doesn't.