Testing

@pleach/core is designed to be tested without a real database or LLM provider account. The @pleach/core/testing subpath drives a real turn from a scripted model stream and covers most agent-level tests; the in-memory storage/checkpointer pair keeps state assertions race-free, and deterministic fingerprints plus the audit ledger handle integration tests.

import { runScriptedTurn, scriptText } from "@pleach/core/testing";
import { MemoryAdapter } from "@pleach/core/sessions";
import { MemorySaver } from "@pleach/core/checkpointing";

Headless turn driver (@pleach/core/testing)

The @pleach/core/testing subpath 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"

The script fixes the model's side of the turn. Everything else is the real runtime: the tools that execute are the ones you registered on it (see Tools), so a test double for a tool is an ordinary defineTool product, not a special executor.

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

createScriptedClient(config: ScriptedClientConfig) is the seam under runScriptedTurn: a minimal client whose every method returns an empty or pass-through value, just enough for the turn body to consume the scripted provider stream. ScriptedClientConfig is { model, provider, chatId?, userId? } — the snapshot its prompts.getGraphConfig() returns. Reach for it only when hand-assembling a driver; runScriptedTurn and createScriptedAdapter cover the common cases.

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);

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);
}

MemoryAdapter + MemorySaver for tests

A SessionRuntime constructed without a storage adapter or a checkpointer defaults to MemoryAdapter and MemorySaver — no env var, no test-only mode. That default is what makes a bare runScriptedTurn need neither a database nor provider credentials.

When you want explicit construction:

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.

What reproduces, and what doesn't

A scripted turn is reproducible by construction, not by a seed: the script is the model's reply, so the same script yields the same result.text and the same event sequence on every run. There is no generated stand-in content to stabilize.

What does not reproduce is timing. ttfbMs and totalMs are wall-clock measurements — assert bounds or presence, never a snapshot.

For the runtime's own non-deterministic reads, SessionRuntimeConfig carries three injectable seams — clock, rng, and idGen. Each defaults to a passthrough (Date.now(), Math.random(), crypto.randomUUID()), so wiring them changes no behavior; supplying your own pins wall-clock, randomness, and minted ids for a test. See Determinism for the end-to-end replay story.

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. collectTurn from @pleach/core/testing does this for you and adds timing; a hand-rolled drain looks like:

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);

When the turn is driven by a real provider rather than a script, prefer assertions on event counts and ordering over assertions on exact deltas — model responses shift across versions.

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/fingerprint";

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 on a runtime, drive it with createScriptedAdapter + collectTurn, and assert on stream events + ledger rows.

CI patterns

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

1. Script the model — runScriptedTurn needs no provider credentials and no network.
2. Let storage default to memory — no database to provision, and the same test runs identically in CI and against staging.
3. Snapshot the ledger, not the stream — token-level deltas shift with provider versions; ledger row shape doesn't.

Where to go next