Verifying behavior

When building systems that others depend on, we often want guarantees that they don't have unintended consequences in the world. For traditional software, we have setups like unit tests and integration tests to give us these guarantees. They help us sleep better at night.

Agentic software is a whole different beast compared to traditional software. It is highly nondeterministic, and its actions can have consequences far beyond the system itself. Traditional ways of testing often fall short in providing enough guarantees for an agent's behaviors. As engineers building these systems, we're left with lingering doubts and half-baked guarantees.

Tardigrade's solution to this problem is property checking using generated actor trajectories.

Good state, bad state

In Tardigrade, the systems we build are defined as state machines. When we describe a system as not having unintended consequences in the world, we mean that our state machine cannot exist in a certain set of unsafe states.

If you find the idea of state machines and possible states too abstract, think of it as a Minesweeper-like board with hidden mines and a goal. Each tile is a state that an actor can step into after performing a certain action.

Translated this way, unintended consequences are paths that our actor should never take. These are the paths that have mines . The mines of your world are specific to the things you build your system around. The constraints that keep our systems from stepping on these mines while pursuing a goal are called safety rules.

Example: a paperclip factory

Imagine we're building an actor that is responsible for operating a paperclip factory. The factory must produce 1,000 paperclips a month. Each batch makes 500 clips and six units of wastewater. The tank can hold ten units.

The actor code would look like this:

import { actor } from "tardie/core"
import { agentMessageMethod, infer, outputValidateOnce, system } from "tardie/agent"
import { inspectWaterTank, producePaperClip, treatWasteWater } from "./factory-tools"

const clippy = actor({
  name: "clippy",
  methods: { message: agentMessageMethod },
  components: [infer([
    system("Produce 1000 paper clips."),
    outputValidateOnce,
    inspectWaterTank,
    producePaperClip,
    treatWasteWater,
  ])],
})

import { Effect } from "effect"
import { tool } from "tardie/agent"

const factorySpec = (name: string) => ({
  name,
  description: name,
  inputSchema: { type: "object" as const, properties: {}, additionalProperties: false },
})

export const inspectWaterTank = tool({
  spec: factorySpec("inspectWaterTank"),
  run: () => Effect.succeed({ action: "inspected" }),
}, "", { name: "inspectWaterTank" })

export const treatWasteWater = tool({
  spec: factorySpec("treatWasteWater"),
  run: () => Effect.succeed({ action: "treated" }),
}, "", { name: "treatWasteWater" })

export const producePaperClip = tool({
  spec: factorySpec("producePaperClip"),
  run: () => Effect.succeed({ action: "produced", wasteAdded: 6 }),
}, "", { name: "producePaperClip" })

import { Schema } from "effect"
import type { Event } from "tardie/core"
import {
  replayProjection,
  type Projection,
} from "tardie/core"

const FactoryResult = Schema.Union([
  Schema.Struct({ action: Schema.Literal("inspected") }),
  Schema.Struct({ action: Schema.Literal("denied"), reason: Schema.String }),
  Schema.Struct({
    action: Schema.Literal("produced"),
    wasteAdded: Schema.Finite,
  }),
  Schema.Struct({ action: Schema.Literal("treated") }),
])
type FactoryResult = typeof FactoryResult.Type

type Tank = {
  capacity: number
  storedWaste: number
  inspected: boolean
  productionWithoutInspection: boolean
}

const tankProjection: Projection<Tank, Tank> = {
  initial: () => ({
    capacity: 10,
    storedWaste: 0,
    inspected: false,
    productionWithoutInspection: false,
  }),
  step: (tank, event) => {
    // ToolReturned records completion; a requested or failed action adds no waste.
    if (event.type !== "ToolReturned" || event.isFailure) return tank
    if (!Schema.is(FactoryResult)(event.result)) return tank

    const result = event.result
    switch (result.action) {
      case "denied":
        return tank
      case "inspected":
        return { ...tank, inspected: true }
      case "produced":
        return {
          ...tank,
          storedWaste: tank.storedWaste + result.wasteAdded,
          productionWithoutInspection:
            tank.productionWithoutInspection || !tank.inspected,
          inspected: false,
        }
      case "treated":
        return {
          ...tank,
          storedWaste: 0,
          inspected: false,
        }
    }
  },
  output: tank => tank,
}

// factoryState folds this prefix of the log through tankProjection.step.
export const factoryState = (events: ReadonlyArray<Event>): Tank =>
  replayProjection(tankProjection, events)

Whenever our actor produces a batch of paperclips, it generates 6 units of wastewater. Our factory tank can only hold 10 units. If our actor continuously produces paperclips without treating the wastewater, it would eventually spill and cause damage to the surrounding area.

The safety rule: prevent a wastewater spill

In this example, the unintended consequence of our system is spilling wastewater. This is an effect of not treating the wastewater before producing more paperclips. To prevent this, we might describe a safety rule for our system as follows:

Inspect the water tank before every production. Treat the wastewater if producing the next batch would cause an overflow.

Here's an example of two paths, one that breaks the rule and one that doesn't:

Finding unsafe trajectories

Generate possible trajectories

To verify if our actor would potentially step into one of the mines, we can generate all possible trajectories of an actor and check them against some rules. Doing so is an example of formal verification. Tools like TLA+ help with this. However, it is computationally very intensive to exhaustively check all possible trajectories of an actor.

A nice workaround is to verify only a sample of trajectories. Tardigrade uses fast-check to do this. Given a set of safety rules, we can have a function that generates a subset of possible actor trajectories using fast-check and finds examples of trajectories that break our safety rules. These are called counterexamples.

Define rules

We define a safety rule as a function that receives the event log and asserts a condition that must always hold:

import type { Event } from "tardie/core"

type Rule = (context: {
  readonly events: ReadonlyArray<Event>
}) => void | boolean

It returns false or throws an assertion error to fail a rule. Return true or finish without throwing to pass. Rules must be synchronous. The checker calls every rule before execution with an empty event log and after every recorded event. A failure identifies the rule and the path that broke it.

Here, factoryState(events) is a projection function that reconstructs the tank state from those events. The first rule checks that production never happened without inspection. The second checks that the accumulated waste fits within the tank.

import { expect } from "bun:test"
import type { ActorInvariants } from "tardie/testing"
import { factoryState } from "./projections"

const rules = {
  inspectBeforeProduction: ({ events }) => {
    const tank = factoryState(events)
    expect(tank.productionWithoutInspection).toBe(false)
  },
  preventOverflow: ({ events }) => {
    // events end at the event currently being checked.
    const tank = factoryState(events)
    expect(tank.storedWaste)
      .toBeLessThanOrEqual(tank.capacity)
  },
} satisfies ActorInvariants

import { Schema } from "effect"
import type { Event } from "tardie/core"
import {
  replayProjection,
  type Projection,
} from "tardie/core"

const FactoryResult = Schema.Union([
  Schema.Struct({ action: Schema.Literal("inspected") }),
  Schema.Struct({ action: Schema.Literal("denied"), reason: Schema.String }),
  Schema.Struct({
    action: Schema.Literal("produced"),
    wasteAdded: Schema.Finite,
  }),
  Schema.Struct({ action: Schema.Literal("treated") }),
])
type FactoryResult = typeof FactoryResult.Type

type Tank = {
  capacity: number
  storedWaste: number
  inspected: boolean
  productionWithoutInspection: boolean
}

const tankProjection: Projection<Tank, Tank> = {
  initial: () => ({
    capacity: 10,
    storedWaste: 0,
    inspected: false,
    productionWithoutInspection: false,
  }),
  step: (tank, event) => {
    // ToolReturned records completion; a requested or failed action adds no waste.
    if (event.type !== "ToolReturned" || event.isFailure) return tank
    if (!Schema.is(FactoryResult)(event.result)) return tank

    const result = event.result
    switch (result.action) {
      case "denied":
        return tank
      case "inspected":
        return { ...tank, inspected: true }
      case "produced":
        return {
          ...tank,
          storedWaste: tank.storedWaste + result.wasteAdded,
          productionWithoutInspection:
            tank.productionWithoutInspection || !tank.inspected,
          inspected: false,
        }
      case "treated":
        return {
          ...tank,
          storedWaste: 0,
          inspected: false,
        }
    }
  },
  output: tank => tank,
}

// factoryState folds this prefix of the log through tankProjection.step.
export const factoryState = (events: ReadonlyArray<Event>): Tank =>
  replayProjection(tankProjection, events)

Try exploring a few paths:

Pass the rules to the checker:

import fc from "fast-check"
import { checkActor } from "tardie/testing"
import { testServices } from "./factory-test-model"

const report = await checkActor(clippy, {
  inputs: fc.record({
    method: fc.constant("message"),
    input: fc.record({
      text: fc.constantFrom("Produce paperclips.", "Make 1,000 paperclips."),
    }),
  }),
  services: context => testServices(context),
  invariants: rules,
  numRuns: 1_000,
  maxSteps: 100,
})

import fc from "fast-check"
import { Effect } from "effect"
import type { ActorCheckContext } from "tardie/testing"
import { testInferenceLayer } from "tardie/agent/testing"
import type { Action } from "tardie/log/events"
import type { InferRequest } from "tardie/agent"

const responseArbitrary = (request: InferRequest): fc.Arbitrary<Action> => {
  const complete = fc.constant<Action>({ kind: "complete", output: "Done." })
  if (request.tools.length === 0) return complete

  const call = fc.constantFrom(...request.tools).map<Action>(tool => ({
    kind: "calls",
    calls: [{
      callId: `call-${request.trajectory.length}`,
      name: tool.name,
      arguments: {},
    }],
  }))
  return fc.oneof(complete, call)
}

export const testServices = ({ generate }: ActorCheckContext) => {
  return testInferenceLayer({
    resolve: (model = { provider: "test", model_id: "factory" }) => ({
      model, contextWindowTokens: 128_000,
    }),
    react: request => Effect.sync(() => generate(responseArbitrary, request)),
  })
}

import { Effect } from "effect"
import { tool } from "tardie/agent"

const factorySpec = (name: string) => ({
  name,
  description: name,
  inputSchema: { type: "object" as const, properties: {}, additionalProperties: false },
})

export const inspectWaterTank = tool({
  spec: factorySpec("inspectWaterTank"),
  run: () => Effect.succeed({ action: "inspected" }),
}, "", { name: "inspectWaterTank" })

export const treatWasteWater = tool({
  spec: factorySpec("treatWasteWater"),
  run: () => Effect.succeed({ action: "treated" }),
}, "", { name: "treatWasteWater" })

export const producePaperClip = tool({
  spec: factorySpec("producePaperClip"),
  run: () => Effect.succeed({ action: "produced", wasteAdded: 6 }),
}, "", { name: "producePaperClip" })

See the Actor checker API for limits, results, and replay.

Read the counterexamples

Unsafe trajectories are referred to as counterexamples to our provided safety rule. The checker throws ActorCheckError with a counterexample, the failing invariant, and its event log. For example, it is possible that the model produces batch A of paperclips and batch B without treating the wastewater.

Prevent unsafe trajectories

Once we've seen the counterexample to our safety rule, we can have conditions within our actor's tools to prevent such paths. In factory-tools.ts, update producePaperClip to reject a batch if the tank wasn't inspected or the batch would cause a wastewater spill.

import { actor } from "tardie/core"
import { agentMessageMethod, infer, outputValidateOnce, system } from "tardie/agent"
import { inspectWaterTank, producePaperClip, treatWasteWater } from "./factory-tools"

const clippy = actor({
  name: "clippy",
  methods: { message: agentMessageMethod },
  components: [infer([
    system("Produce 1000 paper clips."),
    outputValidateOnce,
    inspectWaterTank,
    producePaperClip,
    treatWasteWater,
  ])],
})

import { Effect } from "effect"
import { tool } from "tardie/agent"
import { factoryState } from "./projections"

const factorySpec = (name: string) => ({
  name,
  description: name,
  inputSchema: { type: "object" as const, properties: {}, additionalProperties: false },
})

export const inspectWaterTank = tool({
  spec: factorySpec("inspectWaterTank"),
  run: () => Effect.succeed({ action: "inspected" }),
}, "", { name: "inspectWaterTank" })

export const treatWasteWater = tool({
  spec: factorySpec("treatWasteWater"),
  run: () => Effect.succeed({ action: "treated" }),
}, "", { name: "treatWasteWater" })

export const producePaperClip = tool({
  spec: factorySpec("producePaperClip"),
  run: (_, { readEvents }) => Effect.gen(function* () {
    const events = yield* readEvents()
    const tank = factoryState(events)
    if (!tank.inspected) {
      return { action: "denied", reason: "Inspect the tank before production." }
    }
    if (tank.storedWaste + 6 > tank.capacity) {
      return { action: "denied", reason: "Treat wastewater before production." }
    }
    return { action: "produced", wasteAdded: 6 }
  }),
}, "", { name: "producePaperClip" })

import { Schema } from "effect"
import type { Event } from "tardie/core"
import {
  replayProjection,
  type Projection,
} from "tardie/core"

const FactoryResult = Schema.Union([
  Schema.Struct({ action: Schema.Literal("inspected") }),
  Schema.Struct({ action: Schema.Literal("denied"), reason: Schema.String }),
  Schema.Struct({
    action: Schema.Literal("produced"),
    wasteAdded: Schema.Finite,
  }),
  Schema.Struct({ action: Schema.Literal("treated") }),
])
type FactoryResult = typeof FactoryResult.Type

type Tank = {
  capacity: number
  storedWaste: number
  inspected: boolean
  productionWithoutInspection: boolean
}

const tankProjection: Projection<Tank, Tank> = {
  initial: () => ({
    capacity: 10,
    storedWaste: 0,
    inspected: false,
    productionWithoutInspection: false,
  }),
  step: (tank, event) => {
    // ToolReturned records completion; a requested or failed action adds no waste.
    if (event.type !== "ToolReturned" || event.isFailure) return tank
    if (!Schema.is(FactoryResult)(event.result)) return tank

    const result = event.result
    switch (result.action) {
      case "denied":
        return tank
      case "inspected":
        return { ...tank, inspected: true }
      case "produced":
        return {
          ...tank,
          storedWaste: tank.storedWaste + result.wasteAdded,
          productionWithoutInspection:
            tank.productionWithoutInspection || !tank.inspected,
          inspected: false,
        }
      case "treated":
        return {
          ...tank,
          storedWaste: 0,
          inspected: false,
        }
    }
  },
  output: tank => tank,
}

// factoryState folds this prefix of the log through tankProjection.step.
export const factoryState = (events: ReadonlyArray<Event>): Tank =>
  replayProjection(tankProjection, events)

The guard reads the log with yield* readEvents() and uses the same factoryState(events) projection as the safety rules. Tools that do not read the log incur no log-read cost. Denied actions leave the tank unchanged. After treatment, Clippy must inspect again before producing another batch. This example generates one tool call at a time; concurrent production would also need to account for pending batches.

Summing up

On this page, we treated unintended consequences as states we don't want a system to reach. Since Tardigrade actors are state machines, we can use fast-check to explore sample trajectories by varying inputs and service responses. Safety rules let us find trajectories that violate our system's invariants.

The paperclip factory is a simple example. For a work assistant, a safety rule could prevent sending a confidential document to an external recipient, even when the model requests it. Generated trajectories can expose sequences of messages and tool calls that bypass that restriction.

More complex systems involve interactions that change with their state. For a procurement assistant, a supplier might change while an order awaits approval. A safety rule could require approval for the current supplier and amount, so approval for an earlier version of the order cannot authorize the new purchase. The state we need to track now includes the relationships between orders, suppliers, and approvals.

Going further

The world can change without the actor doing anything. A sensor or API can act as an oracle: record its observations in the event log, then combine them with the actor's actions in a projection. Safety rules can now use what the actor knows about the world.

We can vary these observations in our tests, including stale readings and missing updates. Our guarantees still depend on how accurately those observations represent the world.

Complete example

This test brings Clippy, the simulated tools, and the safety rules together. It finds an unsafe trajectory until the production guard is added.

import { expect, test } from "bun:test"
import fc from "fast-check"
import { actor } from "tardie/core"
import { agentMessageMethod, infer, outputValidateOnce, system } from "tardie/agent"
import { checkActor, type ActorInvariants } from "tardie/testing"
import { inspectWaterTank, producePaperClip, treatWasteWater } from "./factory-tools"
import { factoryState } from "./projections"
import { testServices } from "./factory-test-model"

const clippy = actor({
  name: "clippy",
  methods: { message: agentMessageMethod },
  components: [infer([
    system("Produce 1000 paper clips."),
    outputValidateOnce,
    inspectWaterTank,
    producePaperClip,
    treatWasteWater,
  ])],
})

const rules = {
  inspectBeforeProduction: ({ events }) => {
    expect(factoryState(events).productionWithoutInspection).toBe(false)
  },
  preventOverflow: ({ events }) => {
    const tank = factoryState(events)
    expect(tank.storedWaste).toBeLessThanOrEqual(tank.capacity)
  },
} satisfies ActorInvariants

test("Clippy respects the factory safety rules", async () => {
  const report = await checkActor(clippy, {
    inputs: fc.record({
      method: fc.constant("message"),
      input: fc.record({
        text: fc.constantFrom("Produce paperclips.", "Make 1,000 paperclips."),
      }),
    }),
    services: testServices,
    invariants: rules,
    numRuns: 1_000,
    maxSteps: 100,
  })

  expect(report.status).toBe("passed")
})

Towards formal verification

Passing these tests means no safety rule failed on the generated trajectories. A different trajectory can still reach an unsafe state. With formal verification, we can exhaustively search through all possible states of a state machine, or prove certain properties by induction. Though Tardigrade doesn't natively support this at the moment, you can easily model your agent state machine in TLA+ to prove certain behaviors. We will explore this in a future guide.