Test agents without a network

This is the documentation for unreleased changes on main. For the released library, read v0.1.

Two rules that shape everything here:

• No test makes a real network call. Every external call goes through an injectable seam: ModelClient, fetch_impl, Sandbox.
• No mocked store. Store tests run against real databases. The in-memory adapter tests other components and never proves the store contract.

FakeModel

from psych_runtime.testing.fake_model import FakeModel

model = (
    FakeModel()
    .turn(text="Checking.", tool_calls=[("lookup_order", {"order_id": "A1"})])
    .turn(text="A1 has shipped.")
)

.turn() takes text, tool_calls, reasoning, usage, cost, finish_reason and chunk_delay_seconds. text and reasoning accept a sequence to script it arriving in chunks.

Failure modes, because the ones you never test are the ones that page you:

FakeModel().stalls(seconds=None)  # a stream that goes quiet
FakeModel().aborts_mid_token(text="partial")  # dies halfway through
FakeModel().raises_transient(status_code=503)  # retryable
FakeModel().raises_permanent(status_code=400)  # not retryable
FakeModel().turn(tool_calls=[("no_such_tool", {})])  # unknown tool name

Script malformed tool calls with a ToolCallScript carrying raw wire arguments, which is how you test what happens when a provider sends unparseable JSON.

A weak fake produces a weak suite, so the fake's own coverage is part of the gate in this repository.

Assert on what the model was sent

model.requests  # every ModelRequest, in order
model.last_request  # the most recent

assert "lookup_order" in {t.name for t in model.last_request.tools}

That is how you check prompt assembly and that the tool set is genuinely recomputed each turn rather than pinned at boot.

Running past the script raises

FakeModelScriptExhausted rather than repeating the last turn. Repeating would let a model stuck in a tool-calling loop look like it eventually produced a clean final answer. The error names how many turns were scripted and which call went past them, so it explains itself without a debugger.

A whole run in a test

import psych_runtime
from psych_runtime.store.memory import InMemoryStore
from psych_runtime.testing.fake_model import FakeModel


async def test_it_looks_the_order_up() -> None:
    model = (
        FakeModel().turn(tool_calls=[("lookup_order", {"order_id": "A1"})]).turn(text="Shipped.")
    )

    async with psych_runtime.session(model, tools=[lookup_order]) as s:
        view = await s.ask(spec, "where is order A1?")

        assert view.text == "Shipped."
        report = await psych_runtime.report(s.store, view.run_id)
        assert [c.tool for c in report.tool_calls] == ["lookup_order"]

Assert through the log, with psych_runtime.report(), psych_runtime.answer() or psych_runtime.records(). A feature that works only in memory is a feature that breaks the moment a Worker dies.

LogBuilder: a log without running anything

For testing a projection, a reducer change or a UI against a log shape that is hard to produce for real.

from psych_runtime.testing.logs import LogBuilder

log = (
    LogBuilder(run_id, scope)
    .admitted()
    .attempt()
    .turn()
    .model_started()
    .model_finished(text="", tool_calls=[...])
    .tool_started("c1", tool="issue_refund")
    .suspended(reason=SuspendReason.APPROVAL)
    .resumed(approved=True)
    .tool_finished("c1")
    .settled()
    .build()
)

It has a method per record type, including abort, enqueued, cancelled, consumed, compacted, spawned, messaged_child, child_finished and attempt(reclaimed=True). Building a contradictory log on purpose is how you test that CorruptLog fires.

McpStubServer

A real MCP server over a real loopback socket, not a mock.

from psych_runtime.testing.mcp_stub import McpStubServer, make_server, wire_tool

It records what it was asked on .requests and .calls, and can be told to stall, break a stream mid-response, demand OAuth, or return each documented error code. make_server and wire_tool assemble a Spec against one.

Set tools_page_size to exercise catalogue pagination and structured_results[tool_name] to return typed result data. Use modern_listener_ids to observe active subscriptions and close_modern_listeners() to prove that a client reconnects its listener.

Set sse_responses=True to answer tools/list as a server-sent event instead of application/json. Both are legal and a client accepts either, but only the event carries a size ceiling, so only this shape can refuse a catalogue for being too large. The stub answered JSON exclusively until a real server's two-megabyte catalogue was reported as an unreachable host, which is the shape of bug a stub that never takes the other path cannot catch.

It lives in the shipped package rather than the test suite because it has two consumers and only one is a test: the playground drives it for the scenarios a person clicks through. Nothing in it imports pytest, so depending on it costs an application nothing.

Contract suites

from psych_runtime.store.contract import StoreContractSuite
from psych_runtime.store.blob_contract import BlobStoreContractSuite
from psych_runtime.sandbox.contract import SandboxContractSuite  # both adapters agree

Subclass and override the one fixture. The suite is the deliverable as much as the adapter: divergent implementations with no shared test are divergent behaviours waiting to be found in production.

SandboxContractSuite is the largest of them, because a sandbox has the most to prove and the least room to differ. Every backend — local subprocess, namespaces, Windows job objects, container, remote — subclasses it with one sandbox fixture and is held to the same observable behaviour: stream separation and ordering, binary and invalid-Unicode output, capture caps, artifact collection and escape attempts, CPU, memory, wall-clock and process limits, cancellation before, during and inside a binding, process-tree cleanup, environment isolation, honest network reporting, and the refusal to return output produced under weaker isolation than was asked for. Cases that need a guarantee the backend reports as UNAVAILABLE skip themselves from describe(), so a skip names a real platform limit rather than hiding one.

Testing code execution without a sandbox

from psych_runtime.testing.sandbox_service import ScriptedSandbox, scripted_result

ScriptedSandbox satisfies the Sandbox protocol and runs nothing: it returns the SandboxResult you scripted for a program, records what it was asked, and replays binding calls you name. Use it to test what your agent does with a result — a failure, a limit, a spilled output — without paying for a process.

from psych_runtime.testing.sandbox_service import SandboxService, ServiceFault

SandboxService is a real local HTTP server speaking the remote-sandbox protocol, backed by whatever Sandbox you hand it. It is how RemoteSandbox is tested: no external provider is ever called, and faults injects rejected auth, a 500, a disconnect after ready, a malformed frame, a hang, or a service that claims weaker isolation than it promised.

Three layers

The e2e layer is the regression gate. Every feature ships with a case that fails when the feature breaks. A green unit suite over a broken e2e case is a broken build. Coverage is not the metric: a feature with no e2e case asserting its behaviour is not done, whatever the line coverage says.

Simulating a crash

Kill the Worker mid-tool-call and start another against the same store. The second reclaims the expired lease, replays the log, settles the dangling call and completes. This works because nothing about a running Run lives in a process, and it is one of the ten things the design says must be true.

Gotchas

• MemoryStore is fine for testing components, not for proving the store contract. Those are different jobs.
• Set a short lease_seconds on the Worker in a crash test, or the test waits out the real one.
• filterwarnings = ["error"] in this repository, so a warning fails a test.
• asyncio_mode = "auto", so async tests need no decorator.