Unit testing

#makeTestContext

makeTestContext(options?) returns an AppContext — the exact ctx a mutation / query / action / tool executor receives at runtime — with the deterministic test doubles added on top. Because it is an AppContext, you pass it straight into a handler: await myHandler(input, makeTestContext({ … })). The acting subject is at ctx.request.subject; ctx.store is the real mixin-wrapped store backed by an in-memory data store (so tenant auto-scoping, soft-delete filtering, audit auto-fill, and the fluent select / update / delete builders behave exactly as in production); ctx.cache is a real in-memory cache whose TTLs honour ctx.clock.

Tables are auto-wired from whatever the test imported (the globally-registered tables), so you never re-declare schema. Seed rows with mockStore.

import { makeTestContext, mockStore } from '@voltro/testing'
import createTodo from '../mutations/todos.create.mutation.server'

test('createTodo inserts a row for the caller’s tenant', async () => {
  const ctx = makeTestContext({
    subject: { type: 'user', id: 'user_1', tenantId: 't1' },
    store:   mockStore({ todos: [] }),
  })

  const row = await createTodo({ tenantId: 't1', title: 'buy milk' }, ctx)

  expect(row.title).toBe('buy milk')
  // The real store middleware auto-stamped audit + tenant columns —
  // no different from a live mutation.
})

The same call tests a tool — a tool's execute / handler takes (input, ctx):

import { makeTestContext, mockStore } from '@voltro/testing'
import searchDocs from '../tools/searchDocs.tool'

test('searchDocs returns matching rows', async () => {
  const ctx = makeTestContext({ store: mockStore({ docs: [{ id: 'd1', body: 'hello world' }] }) })
  const out = await searchDocs({ query: 'hello' }, ctx)
  expect(out).toHaveLength(1)
})

#Options

MakeTestContextOptions:

Field	Default	Meaning
subject	anonymous, no tenant	The acting Subject.
tables	all registered	Tables to wire into the store's schema.
store	{}	Seed rows keyed by table name — use mockStore({...}).
clockStart	2026-01-01T00:00:00Z	Frozen start instant for ctx.clock.
ai	—	Injected AI mock (a mockAi({...}) value).
llmResponses	[]	Queued responses for the bundled ctx.llm (MockLLM).
env	ambient process.env	Env values sealed into the boot snapshot so handler code reading getSecret('X') / serverEnv.X resolves under test. Merged over process.env (these win).
relations	—	relations() specs to register for this context — the boot sweep's stand-in. See Eager loads under test.
rowFilter	the registered filter	A row filter for this context only, instead of the process-global setRowFilter(...). See Row-level security under test.

The returned TestContext carries { clock, email, llm, ai?, request, cache, store, withSubject, withTenant } — read the acting subject at ctx.request.subject and the in-memory cache at ctx.cache.

#invoke — run a handler through its guards, its input Schema, and its transaction

Calling an executor directly skips the hops the real dispatcher runs first: the descriptor's guards, the input-Schema decode, and — for a mutation — the transaction. So a test can feed the handler a value the wire would reject, can exercise a handler the caller was never authorized to reach, and can leave half-written state behind that production would have rolled back. invoke(descriptor, executor, rawInput, ctx) closes them — it decodes rawInput through descriptor.input, enforces descriptor.guards against ctx.request.subject, then calls the executor with the decoded value, wrapping a mutation in a real store transaction. It also runs the hops around those: the plugin interceptor chain, the post-commit (afterCommit) drain, and the deadlock replay.

import { invoke, makeTestContext } from '@voltro/testing'
import { createNote } from './notes.mutation'          // the descriptor
import { createNoteHandler } from './notes.mutation.server' // the executor

const ctx = makeTestContext({ subject: { type: 'user', id: 'u1', tenantId: 't1' } })

// valid input decodes, then the handler runs
const note = await invoke(createNote, createNoteHandler, { title: 'hi' }, ctx)

// a bad shape rejects at the decode — the handler never runs
await expect(invoke(createNote, createNoteHandler, { title: 42 }, ctx)).rejects.toThrow()

#Effect-mode handlers run too

A handler may be written async or as an Effect — the framework's contract is "your choice, per handler", and the dispatcher runs both. invoke makes the same test, so an Effect-returning executor is executed, and invoke resolves to its success value (typed as that value, not as the Effect):

export const publishNote = (input: { id: string }, ctx: AppContext) =>
  Effect.gen(function* () {
    const store = yield* EffectStore
    yield* store.update('notes', input.id, { published: true })
    return 'published'
  })

const out = await invoke(publish, publishNote, { id: 'n1' }, ctx)
expect(out).toBe('published')     // the value — not an un-run Effect

A failure on the typed error channel rejects with that error, exactly as an async handler's throw does — so the same assertion works for either mode:

await expect(invoke(publish, publishNote, { id: 'gone' }, ctx))
  .rejects.toMatchObject({ _tag: 'NoteNotFound' })

EffectStore and SubjectService are provided over the context the handler is actually given — inside a mutation that is the transactional one, so an Effect handler's writes roll back with everything else. Guards, the input decode, the transaction, the deadlock replay, afterCommit and the plugin interceptors all wrap the Effect form identically. An app's own layers: and the aggregate registry are not provided: those are boot injections the harness has no access to.

#Guards are enforced

An unauthorized caller is refused with the typed ScopeError — the same error a client would receive — before the handler runs:

// the descriptor declares `guards: [scope('notes:write')]`
const outsider = makeTestContext({
  subject: { type: 'user', id: 'u2', tenantId: 't1', scopes: [] },
})

await expect(invoke(createNote, createNoteHandler, { title: 'hi' }, outsider))
  .rejects.toMatchObject({ _tag: 'ScopeError' })

What this covers:

• Scope guards — checked against ctx.request.subject.
• Resource-scoped guards — resolved through whatever setResourceScopeResolver the test registered.
• Relationship / policy guards — resolved through whatever tuple source the test registered. With none registered they DENY, exactly as in production: an authorization question nobody can answer is a refusal.

The full order invoke runs, which is production's:

decode → plugin interceptors( guards → [transaction → handler] → afterCommit )

The decode is first because in production it happens at the wire (@effect/rpc), before any runner is reached — so a malformed payload rejects with a parse error even for a caller no guard would have let through. Guards then run before the transaction opens, and inside the plugin chain, so an rbac-style plugin that publishes a subject's role scopes from its interceptor has already run when the guard is checked. A resource extractor therefore sees the decoded input, exactly as on the dispatch spine.

#Mutations run in a real transaction

A mutation invoked through invoke runs inside store.transactional(...) — the store's own method, the same one the serve pipeline calls. So "the mutation failed, therefore nothing was written" is something you can assert here rather than deferring to an e2e test:

// the handler inserts, then throws on the second step
await expect(invoke(createNote, failingHandler, { title: 'hi' }, ctx)).rejects.toThrow()

// nothing survived — the rollback is the store's, not a copy-aside restore
expect(await ctx.store.select('notes').all()).toHaveLength(0)

The rollback is real: the in-memory data store's transactional view keeps a private overlay and discards it on a throw (buffered change events drain only on commit). Nothing in the harness copies rows aside and puts them back.

Which procedures get wrapped comes off descriptor.kind, which every defineQuery / defineMutation / defineAction / defineStream stamps — you never declare it, and so can't declare it wrongly:

• kind: 'mutation' → wrapped.
• Queries and actions → not wrapped. That mirrors production rather than omitting something: an action deliberately runs outside a transaction because it performs external I/O that cannot be rolled back.
• A hand-rolled descriptor object with no kind → not wrapped. There is nothing to key off, and wrapping everything would give an action the wrong semantics.

The context the handler receives is re-derived over that transaction: txCtx.store, its withSubject / withTenant re-scopers, and its load / loadMany batcher all read and write through the same transaction, so a handler can't accidentally escape it mid-mutation.

#runInStoreTransaction

The same wrap is exported for tests that want it around something other than an invoke call:

import { runInStoreTransaction, makeTestContext } from '@voltro/testing'

const ctx = makeTestContext({ store: mockStore({ notes: [] }) })

await expect(
  runInStoreTransaction(ctx, async (txCtx) => {
    await txCtx.store.insert('notes', { title: 'a' })
    throw new Error('boom')
  }),
).rejects.toThrow()

expect(await ctx.store.select('notes').all()).toHaveLength(0)

Nested calls are not supported — the in-memory store rejects a nested transaction, exactly as it does at runtime.

#afterCommit and ctx.outbox

Post-commit work runs, after the commit and never after a rollback — which is the entire contract of afterCommit: the side effect happens if and only if the write did. ctx.outbox.enqueue(...) schedules its delivery nudge through exactly that hook, and the enqueue itself is atomic with your domain write, so a throw loses both. Read the nudges a call produced with outboxNudgesOf(ctx):

await invoke(createNote, async (input, c) => {
  await c.store.insert('notes', { title: input.title })
  await c.outbox.enqueue('note.created', { title: input.title })
  return 'ok'
}, { title: 'hi' }, ctx)

expect(outboxNudgesOf(ctx)).toHaveLength(1)

#Plugin interceptors

Pass the plugins to the context and their rpc interceptors wrap every invoke on it — the same list you declare in app.config.ts, composed the same way (first listed is outermost):

const ctx = makeTestContext({ subject, plugins: [auditPlugin, rbacPlugin] })

// the plugin's interceptMutation now wraps this call
await invoke(createNote, createNoteHandler, { title: 'hi' }, ctx)

The harness takes the plugin objects, not a bare function, so it makes the same kind-selection production does: interceptMutation for a mutation, interceptQuery for a query, interceptAction for an action. A hook filed under the wrong name silently never fires here — exactly as it silently never fires in production, which is the bug worth catching.

Your interceptor receives what it receives at runtime: { tag, kind, input, subject, traceId, spanId? }, with input already decoded. It can short-circuit (return a different Effect and the handler never runs), transform the result, or tap the error channel. Typed errors round-trip unwrapped — a plugin in the chain does not turn your handler's NoteNotFound into an opaque defect.

Two ordering properties you can assert directly, because they are the ones that bite:

// 1. The chain wraps the GUARDS — an interceptor sees the ScopeError.
//    (This is what lets an rbac plugin publish scopes before the check.)
// 2. The chain runs OUTSIDE the transaction — an interceptor that throws
//    AFTER the commit does not roll the mutation back:
const plugin = definePlugin({
  name: '@acme/audit',
  framework: '^1.0.0',
  interceptMutation: (next) =>
    next.pipe(Effect.flatMap(() => Effect.fail(new Error('audit sink down')))),
})

const ctx = makeTestContext({ subject, plugins: [plugin] })
await expect(invoke(createNote, createNoteHandler, { title: 'hi' }, ctx)).rejects.toThrow()

// the write is still there — post-only side effects must not undo the mutation
expect(await ctx.store.select('notes').all()).toHaveLength(1)

Only the rpc interceptors are wired. Lifecycle hooks (onActivate), schema, routes and dashboard mounts are boot concerns with no meaning for a single handler call, and are ignored. rpcInterceptorFor(ctx, 'mutation') returns the composed chain if you want to assert on it without invoking.

#The deadlock replay

Production replays a mutation whose transaction lost a deadlock — under concurrency the contract is "the victim retries", not "the write fails". invoke does the same, using the runtime's own classifier (mysql/maria errnos, the pg/mssql serialization SQLSTATEs, walked down the cause chain), so what counts as deadlock-shaped has one definition:

let attempts = 0
await invoke(createNote, async (input, c) => {
  attempts++
  await c.store.insert('notes', { title: input.title })
  if (attempts === 1) throw Object.assign(new Error('Deadlock found'), { errno: 1213 })
  return 'ok'
}, { title: 'hi' }, ctx)

expect(attempts).toBe(2)
expect(await ctx.store.select('notes').all()).toHaveLength(1) // only the surviving attempt

An ordinary failure is not replayed — it throws on the first try, so a plain bug never runs your handler three times.

The property that makes a replay safe is that each attempt starts clean: the rolled-back attempt's writes are gone (the transaction), and so is its queued post-commit work. Without that reset a replayed mutation would fire the outbox nudges of writes that never landed:

// attempt 1 enqueues then deadlocks; attempt 2 enqueues and commits
expect(outboxNudgesOf(ctx)).toHaveLength(1) // one, not two

Retries are always on and carry no backoff here. Production's jitter exists to de-correlate concurrent lock victims; a unit harness has neither concurrency nor a lock manager, so a sleep would only cost wall-clock (and stall a suite on fake timers). The retry semantics — attempt budget, transient classification, per-attempt reset — are what's reproduced. There is no opt-out, and it cannot mask a real failure: a handler that fails deterministically fails identically after the last attempt.

#What invoke still does NOT cover

• Transport concerns — connection info, rate limiting, the tenant header. Those are properties of the HTTP hop, not of the procedure; faking them here would only assert against the fake.
• Undo capture and the metrics sample — optional injections the serve entrypoint makes. Their absence changes nothing a handler can observe.

#Completing a fixture row — fixtureRow

ctx.store is the real mixin-wrapped store, so store.insert runs the same required-column validation production does: a payload that omits a NOT-NULL, no-default, non-auto-stamped column throws TableValidationFailed naming the column. That is deliberate — the test store is not laxer than Postgres, so a fixture can't pass on a row the database would reject. But it means a lean fixture (insert('journal_entries', { id }), an omitted required FK) now fails.

fixtureRow(table, overrides) completes the row for you. It fills every column the validator would flag with a schema-typed placeholder, then merges your overrides on top — so you write only the columns the test cares about and the rest are made valid:

import { fixtureRow, makeTestContext } from '@voltro/testing'
import { journalEntries } from '../schema/journal'

const ctx = makeTestContext({ subject, store: mockStore({ journal_entries: [] }) })

await ctx.store.insert('journal_entries', fixtureRow(journalEntries, {
  amount: '100.00',          // the columns THIS test asserts on
}))                          // entryNumber (unique), postedAt, … auto-filled

What it fills and what it leaves alone:

• Fills each required column with a value of the right shape — a oneOf column takes its first allowed value, a unique column gets a distinct value per call (so two fixtures don't collide on the key), timestamp / date get a fixed epoch, decimal / bigint a numeric string.
• Leaves out exactly what a caller may legitimately omit: nullable columns, defaulted columns, and the framework auto-stamped set (id, tenantId, createdAt, updatedAt, createdBy, updatedBy, deletedAt, deletedBy) — the store stamps those from the subject and the table scheme.
• Refuses to guess a structured type (json, bytes, vector, array, interval, raw): it throws naming the column and telling you to pass it explicitly, rather than inventing a value that fails the codec obscurely.
• Your override always wins — including an explicit null. Don't assert on a synthesized placeholder; override anything the test inspects.

There is no flag to turn the validation off — a test store that accepts rows production rejects is a fake testing itself. fixtureRow is a runtime filler for the loose store.insert(name, row) path. For compile-time payload typing (a missing required column caught as a type error at the call), use insertRow / upsertRow from @voltro/database, which check against InferInsertRow<T>.

#Subject + tenant re-scoping

withSubject and withTenant re-scope to a different principal for one block, sharing the same underlying data — so cross-subject reads exercise real tenant scoping, not a closure stub. This is how you prove isolation: write as one tenant, then assert another tenant can't see the row.

const ctx = makeTestContext({
  subject: { type: 'user', id: 'u1', tenantId: 't1' },
  store:   mockStore({ todos: [] }),
})

await ctx.store.insert('todos', { title: 'private to t1' })

await ctx.withTenant('t2', async (ctx2) => {
  const rows = await ctx2.store.select('todos').all()
  expect(rows).toHaveLength(0)   // t2 sees none of t1's rows
})

withSubject(subject, fn) swaps the full identity; withTenant(tenantId, fn) keeps the current subject and only changes the tenant. Both return a Promise of whatever fn returns.

#Asserting on hidden rows

ctx.store applies the same scoping a handler gets — the tenant filter plus deletedAt IS NULL. To assert on rows the scope hides (that a mutation soft-deleted a row, or wrote into another tenant), read past the scope with .unscoped() (drops the tenant filter) and .withDeleted() (includes soft-deleted rows) — both typed, no cast:

await deleteNote({ id: 'n1' }, ctx)                 // soft-delete
expect(await ctx.store.select('notes').all()).toHaveLength(0)   // hidden from normal reads

const raw = await ctx.store.select('notes').unscoped().withDeleted().all()
expect(raw[0]?.deletedAt).not.toBeNull()            // …but still there, tombstoned

#Eager loads under test

relations() is pure — it returns a spec, it does not register one. In production voltro dev discovers every *.relations.ts and registers what it exports; a unit test runs no boot, so importing the module registers nothing and the first .with({ … }) fails with "no relations registered". Hand the specs to the context instead:

import { teamRelations } from '../db/teams.relations'

const ctx = makeTestContext({
  relations: [teamRelations],
  store: mockStore({ teams: [{ id: 't1' }], members: [{ id: 'm1', teamId: 't1' }] }),
})

const rows = await ctx.store.select('teams').with({ members: true }).all()
expect(rows[0].members).toHaveLength(1)

The relations registry is process-global, so the option replaces it with exactly the specs you pass rather than adding to it. That is what keeps two makeTestContext({ relations: [...] }) calls in one file independent — additive registration would throw duplicate relation on a re-registered spec and would carry the first test's relations into the second. Omitting the option leaves the registry untouched.

#Row-level security under test

ctx.store applies the app's row filter for the context's subject: registered with setRowFilter(...), resolved once per context, AND-merged into every read. Both read paths are covered (the fluent builders and store.query(descriptor)), .unscoped() does not bypass it — that opts out of tenant isolation, not of authorization — and a system subject bypasses it, exactly as at runtime.

const ctx = makeTestContext({ subject: alice, store: mockStore({ tickets: seed }), rowFilter: ownTickets })

const rows = await ctx.store.select('tickets').all()
expect(rows.map((r) => r.id)).not.toContain('bobs-ticket')  // the rule, asserted

Pass rowFilter: — as above — to scope a filter to this context only. setRowFilter is process-global: registered in one test it silently constrains every later test in the same worker, and a forgotten afterEach surfaces as a failure in an unrelated file. Either way the resolution is the runtime's own, so the retry schedule, the system bypass and the onLoadError policy behave identically: a filter whose load fails refuses the read (with RowFilterUnavailable, or zero rows under onLoadError: 'deny') rather than quietly returning everything.

#Filters over SHARED resources

A filter over rows the user owns needs no database — the subject carries the id. A filter over rows shared with the user must read a membership table, and load is Effect<Ctx, unknown> with R = never, so it cannot yield* a store service. Its only route is runAsSystem, and makeTestContext wires its seeded store into that, so this resolves under test exactly as it does at runtime:

const sharedLists: RowFilter<{ listIds: ReadonlyArray<string> }> = {
  load: (subject) =>
    Effect.promise(() =>
      runAsSystem(async (sys) => {
        const rows = await sys.store.select('listMembers').where('userId', String(subject.id)).all()
        return { listIds: rows.map((r) => String(r.listId)) }
      }),
    ),
  predicate: (ctx, table) => (table === 'lists' ? inSet('id', ctx.listIds) : undefined),
}

const bob = makeTestContext({ subject: bobSubject, store: mockStore(seed), rowFilter: sharedLists })
expect((await bob.store.select('lists').all()).map((r) => r.id)).not.toContain('alices-list')

makeTestContext registers the raw data store + schema registry, so nothing is double-wrapped (runAsSystem applies the system-subject mixin wrap itself), and it is the outer store rather than a transactional view — matching production, where a runAsSystem block inside a mutation reads outside that mutation's transaction.

The handle is a process global, and the harness handles that in two layers. Registration is last-wins, so a bare await runAsSystem(...) written directly in a test also resolves. On top of that, each context re-points the handle at its own store for the span of its load and restores the previous value — necessary because two makeTestContext calls allocate two separate in-memory stores even from one seed object, so filter resolution must not depend on build order. Nothing leaks across test files (vitest gives each file a fresh module registry). Only the harness registers: an app that registers no handle still gets runAsSystem: no data store available, unchanged. If a test in the same file needs that refusal back, call clearSystemStoreHandle() from @voltro/runtime.

#The deterministic mocks

#ctx.clock — MockClock

A frozen clock you advance by hand. No real sleeping.

ctx.clock.now()            // current instant in ms
ctx.clock.date()           // current instant as a Date
ctx.clock.advance('15m')   // jump forward — ms | s | m | h | d, or a number of ms
ctx.clock.advance(900_000) // same, in milliseconds

#ctx.email — MockEmail

Captures outgoing email for assertion.

ctx.email.sent                 // SentEmail[] — every send, in order
ctx.email.lastTo('a@b.com')    // the most recent email to that address, or undefined
ctx.email.clear()              // reset between cases

#ctx.llm — MockLLM

A queue of canned model responses plus a record of every call. Each MockResponse is one of { text }, { toolCall: { name, input } }, or { error: { code, message? } }.

const ctx = makeTestContext({
  llmResponses: [{ text: 'first' }, { toolCall: { name: 'search', input: { q: 'x' } } }],
})

// inside the handler under test, the model consumes the queue in order;
// afterwards:
expect(ctx.llm.calls).toHaveLength(2)   // each prompt that triggered a call
expect(ctx.llm.remaining()).toBe(0)     // queue drained

MockLLM throws MockLLM: no more responses queued if the code under test asks for one more response than you queued — a useful assertion that the loop ran exactly as many turns as expected.

#Testing agents and the model loop

ctx.llm is the low-level queue. For a full agent / model-loop test, install a deterministic provider with mockAi / useMockAi from @voltro/ai/test — the AI package owns the full mock; @voltro/testing only types the injected MockAi slot on makeTestContext({ ai }).

import { useMockAi } from '@voltro/ai/test'

const mock = useMockAi({
  stream: ['Hello, ', 'world.'],                  // token deltas
  turns: [                                        // scripted tool loop
    { toolCalls: [{ name: 'searchDocs', input: { query: 'x' } }] },
    { text: 'Based on the docs.' },
  ],
})
// … run the agent …
mock.reset()

mockAi({...}) returns a MockAi value you pass as makeTestContext({ ai: mockAi({...}) }); throwNTimes(n, value) exercises retry paths. See AI → Providers for the full surface.