Skip to content

agentkit.core.testing

Test doubles for every Protocol — FakeLLM, FakeFetch, FakeSearch, FakeMemory, FakeTool, FakeCompactor, FakeGrounder, FakeClock — plus the make_test_ctx() builder that wires them into a RunContext.

Two doubles are not port doubles, because the CLI cognitions are not behind a port: FakeClaudeCli and FakeCodexCli stand in for the subprocess itself (with the shared CliRun / CliStderr / CliInvocation value types, and codex_turn() to build one well-formed Codex turn), so a test still runs the real parsing, event mapping and meter charging.

Canonical test utilities for agentkit consumers.

What lives here: - make_test_ctx(...) — factory that builds a REAL RunContext for tests, wired with the FakeLLM-backed Invoker + noop Trace/Observer defaults. Suitable for any test that needs a real ctx.

What lives under fakes/: - FakeCtx — minimum Ctx that RECORDS spans (different from agentkit.core.runtime.NullCtx, which records nothing). Use when a test needs to assert on what spans were opened. - FakeLLM, FakeClock, FakeFetch, FakeSearch — port doubles. - FakeGrounder, FakeCompactor — capability doubles for RequestBuilder tests. - FakeClaudeCli / FakeCodexCli (+ the shared CliRun, CliStderr, CliInvocation, and codex_turn for building a well-formed Codex turn) — the two doubles that are NOT port doubles, because the claude and codex CLIs are not behind a port. Each replaces the subprocess, replaying recorded or scripted JSON so the cognition's own parsing, event mapping and meter charging still run.

Naming convention: Fake* is the standard for test doubles. Null* / Noop* are production-grade null-object patterns; they live in their respective runtime/kernel modules, NOT here.

CliInvocation dataclass 

CliInvocation(
    argv: tuple[str, ...],
    cwd: str | None = None,
    env: Mapping[str, str] = dict(),
    stdin: bytes = b"",
    start_new_session: bool = False,
    lines_read: int = 0,
    terminated: bool = False,
    killed: bool = False,
    returncode: int | None = None,
)

What one spawn was asked to do, and what happened to it.

Records argv/cwd/env so a flags test can assert on them without the process-wide create_subprocess_exec patch, and stdin so a session test can assert on what the cognition wrote — which is otherwise invisible, because a session's input side never reaches the filesystem or the event stream.

CliRun dataclass 

CliRun(
    stdout: bytes = b"",
    stderr: bytes = b"",
    returncode: int = 0,
    interleaved_stderr: tuple[tuple[int, bytes], ...] = (),
)

One spawn of a CLI, whole: its stdout, its stderr, its exit code.

Frozen because it is the recording. Two doubles over one CliRun must not be able to affect each other, and the cheapest way to guarantee that is for the shared thing to be immutable and the per-spawn cursor to live in the iterator.

of classmethod 

of(
    payloads: Iterable[
        Mapping[str, Any] | CliStderr | str | bytes
    ],
    *,
    stderr: bytes = b"",
    returncode: int = 0,
) -> CliRun

Assemble a run from payloads.

A Mapping is serialised as one JSON line — the ordinary case. A str or bytes is written verbatim, newline and all, which is the whole malformed-session story in one rule: '{"type":"result",' is a process killed mid-object, '\n' is a blank warm-up line, 'warning: ...\n' is a plain-text diagnostic on stdout. No truncate= knob, no corrupt= flag; every shape a real binary can emit is already expressible, and a knob per shape would only enumerate the ones someone thought of.

CliStderr dataclass 

CliStderr(data: bytes)

A chunk the CLI wrote to stderr at this point in its stdout stream.

Positional, not a lump attached to the run, because a cognition reads stderr exactly once — at the end, after stdout hits EOF. A diagnostic emitted in the middle of a long run therefore has to survive until then, and "did it?" is a real question: stderr is the only channel that explains a bare cli_exit_1 to an operator. Placing chunks in the stream is what lets a test ask it.

FakeClaudeCli 

FakeClaudeCli(
    runs: Sequence[CliRun] = (),
    *,
    repeat_last: bool = False,
)

Deterministic offline claude CLI. Pass it as spawn= to a :class:~agentkit.agents.cognition.ClaudeCliCognition.

Holds an ordered list of :class:CliRun s — one per spawn, not per turn, because that is the unit the CLI actually has: drive() spawns once per run, while a session spawns once and serves many turns off the same stdout. Each spawn consumes the next run; asking for one the recording does not have raises :class:~agentkit.core.testing.ScriptExhausted, for the reason that class exists — a recording is a claim about how many times the CLI is invoked, and code that invokes it more has a defect (a retry that should not have fired, a loop that should have stopped) that replaying the last run would answer with a plausible, wrong result.

repeat_last=True replays the final run forever, for the test that deliberately drives an unbounded number of spawns.

remaining property 

remaining: int

Runs not yet spawned. Negative is impossible; zero after the last spawn is the normal end of a fully-consumed recording.

replay classmethod 

replay(
    path: Path | str,
    *more: Path | str,
    repeat_last: bool = False,
) -> FakeClaudeCli

Replay recorded stream-json session file(s), one spawn per file.

The bytes are read now, once, and held immutably — not streamed off disk per spawn. Two reasons, and neither is performance: a recording that vanishes or changes between construction and the spawn three layers down would surface as tests.evals["error"] inside the cognition's except BaseException, where a missing fixture is nearly unreadable; and "replays byte-identically twice" is trivially true of an immutable buffer and merely probable of a file. Streaming happens on the other side — the buffer is handed to the consumer one line at a time.

script classmethod 

script(
    payloads: Iterable[
        Mapping[str, Any] | CliStderr | str | bytes
    ],
    *,
    stderr: bytes = b"",
    returncode: int = 0,
    repeat_last: bool = False,
) -> FakeClaudeCli

Assemble ONE spawn's output from payloads — the case nobody recorded.

One spawn rather than one-per-payload because that is what the shapes this is for need: a refusal, a session limit and a malformed final answer are all a single CLI invocation that went wrong. Several spawns with different exit codes are FakeClaudeCli([CliRun.of(...), ...]).

See :meth:CliRun.of for how a payload becomes bytes — in particular that a raw str/bytes is written verbatim, which is how a truncated or non-JSON line is built.

FakeClock 

FakeClock(start: float = 0.0)

Deterministic offline ClockPort. start is the initial unix timestamp; now() returns the current virtual time; sleep(s) advances it by s (no real wall-clock wait).

advance 

advance(seconds: float) -> None

Test-only escape hatch: move the clock forward without recording a sleep call (useful for simulating external time passing between agent steps).

FakeCodexCli 

FakeCodexCli(
    runs: Sequence[CliRun] = (),
    *,
    repeat_last: bool = False,
)

Deterministic offline codex CLI. Pass it as spawn= to a :class:~agentkit.agents.cognition.CodexCliCognition.

Holds an ordered list of :class:~agentkit.core.testing.fakes.CliRun s — one per spawn. That unit is worth being precise about, because it differs from the Claude double's in the way that matters for writing tests: codex exec is one-shot, so a spawn is a TURN. drive() consumes one run; a three-turn :class:~agentkit.agents.cognition.CodexCliSession consumes three.

Each spawn consumes the next run; asking for one the recording does not have raises :class:~agentkit.core.testing.ScriptExhausted. repeat_last=True replays the final run forever, which is the shape a session test that does not care how many turns it takes wants.

remaining property 

remaining: int

Runs not yet spawned. Negative is impossible; zero after the last spawn is the normal end of a fully-consumed recording.

replay classmethod 

replay(
    path: Path | str,
    *more: Path | str,
    repeat_last: bool = False,
) -> FakeCodexCli

Replay recorded JSONL session file(s), one spawn per file.

The bytes are read now, once, and held immutably — not streamed off disk per spawn. A recording that vanished or changed between construction and the spawn three layers down would surface as tests.evals["error"] inside the cognition's except BaseException, where a missing fixture is nearly unreadable; and "replays byte-identically twice" is trivially true of an immutable buffer and merely probable of a file. Streaming happens on the other side — the buffer is handed to the consumer one line at a time.

script classmethod 

script(
    payloads: Iterable[
        Mapping[str, Any] | CliStderr | str | bytes
    ],
    *,
    stderr: bytes = b"",
    returncode: int = 0,
    repeat_last: bool = False,
) -> FakeCodexCli

Assemble ONE spawn's output from payloads — the case nobody recorded.

One spawn rather than one-per-payload because that is what the shapes this is for need: a refusal, a stream error and a malformed final answer are all a single CLI invocation that went wrong. Several spawns with different exit codes are FakeCodexCli([CliRun.of(...), ...]).

See :meth:~agentkit.core.testing.fakes.CliRun.of for how a payload becomes bytes — in particular that a raw str/bytes is written verbatim, which is how a truncated or non-JSON line is built.

answering classmethod 

answering(
    *texts: str, thread_id: str = DEFAULT_THREAD_ID
) -> FakeCodexCli

One spawn per text, each a complete well-formed turn saying it.

The shorthand for a multi-turn session test, where the conversation's CONTENT is not what is under test — the argv, the thread id, the stop reason or the metering is — and five hand-written payloads per turn are five chances to write a stream the binary would not emit::

cli = FakeCodexCli.answering("first", "second", "third")

Every turn re-announces the same thread_id, which is what a real resumed run does; a session that failed to thread its id would show up as a missing resume in invocations[1].argv, not as a mismatched id here.

argv 

argv(index: int = -1) -> tuple[str, ...]

One spawn's argv, defaulting to the most recent.

A convenience with a purpose: the flag assertions in this cognition's tests are all "what did spawn N ask for", and cli.invocations[-1].argv reads as plumbing at every one of them.

FakeCompactor dataclass 

FakeCompactor(
    sentinel: str = "[COMPACTED]", called: bool = False
)

RequestBuilder test helper. Replaces every transcript with a single sentinel message so a test can detect that compaction ran.

FakeCtx 

FakeCtx(*, scope: Scope | None = None)

Minimal Ctx that RECORDS spans + observations.

Use when a test needs to assert on what spans were opened (e.g., RequestBuilder stamps agentkit.prompt.version — a test for that contract needs a tracer that captures attributes). Distinct from agentkit.core.runtime.NullCtx: NullCtx absorbs operations and records nothing; FakeCtx records spans for assertion. Both are valid — pick based on whether you need to assert on what was recorded.

tracer.spans is also exposed as ctx.spans for tests that fold the tracer onto the ctx itself. Either access form works.

spans property 

spans: list[tuple[str, str, dict[str, Any]]]

Convenience alias mirroring (name, kind, attrs) tuples on the ctx itself, so tests can assert on ctx.spans directly instead of reaching through ctx.trace.spans.

FakeFetch 

FakeFetch(fixtures: dict[str, FetchResponse])

Deterministic offline FetchPort. fixtures is {url: FetchResponse}; unknown URLs raise KeyError(url) so tests fail loud instead of silently falling back to a stub response.

FakeGrounder dataclass 

FakeGrounder(
    block: str = "", calls: list[tuple[Any, str]] = list()
)

RequestBuilder test helper. Records (ctx, task) calls; returns a canned block (empty string allowed to test the skip path).

FakeLLM 

FakeLLM(
    responses: dict[str, str]
    | Callable[..., str]
    | str = "{}",
    *,
    usage: Usage | None = None,
    fail_times: int = 0,
    fail_exc: BaseException | None = None,
    delay: float = 0.0,
    turns: list[Turn] | None = None,
    repeat_last: bool = False,
)

Deterministic offline LLM. dict {substring_of_user: content} | callable | str for single answers; FakeLLM.script([Turn(...), …]) replays a multi-step tool loop across chat calls.

The two forms differ ON PURPOSE once they run out of material:

  • the single-answer forms (str / dict / callable) have nothing to run out of — they are a rule for turning a request into a reply, so call 1 and call 100 both answer. Repeating IS their meaning.
  • a script is a finite, ordered claim about how many turns the run takes. Asking for turn N+1 contradicts the claim, so it raises ScriptExhausted. Pass repeat_last=True for the deliberate never-terminating script — a single tool-call Turn used to drive a loop until max_iterations or a budget cuts it.

script classmethod 

script(
    turns: list[Turn],
    *,
    repeat_last: bool = False,
    **kw: Any,
) -> FakeLLM

Replay turns in order, one per chat/stream call, then raise ScriptExhausted.

repeat_last=True replays the final turn forever instead. Spelled out as a real keyword rather than left to **kw so it shows up in a signature and an IDE completion — it is the escape hatch a reader needs at the moment the exception fires, and **kw hides it.

stream async 

stream(
    *,
    messages: list[Message],
    model: str,
    tools: list[ToolSchema] | None = None,
    response_format: dict[str, Any] | None = None,
    temperature: float = 0.0,
    max_tokens: int | None = None,
    cache_hint: Any = None,
) -> AsyncIterator[Delta]

The streaming primitive: emits content in small chunks then a terminal delta carrying usage/finish/tool_calls. _enter() raises BEFORE the first delta, so it's a pre-stream failure the retry middleware can re-invoke (honoring fail_times).

FakeMemory dataclass 

FakeMemory(
    name: str = "fake_memory",
    responses: dict[str, list[MemoryItem]] = dict(),
    default: list[MemoryItem] = list(),
    items: list[MemoryItem] | None = None,
)

A MemorySource-Protocol-conforming fake.

Two modes:

  • responses={"q": [MemoryItem(...), ...], ...} — per-query lookup. Misses fall through to default (defaults to []).
  • items=[MemoryItem(...), ...] — static fixture mode. query ignores the query string entirely and returns the first k items regardless. Useful when a test doesn't care about retrieval semantics, only that the agent received some items.

Every query is recorded as (query, k, where) on self.queries; every write batch is appended to self.writes. Lets tests assert mem.queries[0] == ("foo", 3, None) or mem.writes == [[MemoryItem("a", "src")]].

FakeSearch 

FakeSearch(
    hits: list[SearchHit] | dict[str, list[SearchHit]],
)

Deterministic offline SearchPort. Pass a list (returned for every query) or a dict {substring_of_query: [SearchHit, ...]} (first substring match wins; unknown queries → []).

FakeTool dataclass 

FakeTool(
    name: str = "fake_tool",
    description: str = "fake tool for unit tests",
    responder: Any = None,
    side_effecting: bool = False,
    requires_approval: bool = False,
    output_schema: dict[str, Any] | None = None,
    schema: ToolSchema = _default_schema(),
)

A Tool-Protocol-conforming fake.

Construct with a static responder value (returned verbatim from every run) OR with a callable responder(args) -> Any that may be sync or async. Any exception the callable raises propagates, letting a test exercise the error path.

Every run call is appended to self.calls as the (copied) args dict, so tests can do assert tool.calls == [{"x": 1}, {"x": 2}].

The default schema is the permissive {"type": "object"} — any JSON object accepted. Override per-test if you want the Protocol's advertised schema to reflect a real shape.

RecordingSpan 

RecordingSpan(name: str, kind: str, attrs: dict[str, Any])

A span that records its name / kind / attributes / events. Drop-in for the production span surface — exposes .set(key, value) and .add_event(name, **fields). Attributes start from the kwargs passed to span() and are mutated by set().

RecordingTracer 

RecordingTracer(
    *,
    exporter: Callable[[RecordingSpan], None] | None = None,
)

In-process TracePort that records every span. Optional exporter callback receives each span object on close (mirrors the OTel SpanExporter shape at the granularity tests care about).

Satisfies the TracePort Protocol structurally: - span(name, kind, **attrs) — context manager yielding a RecordingSpan - current_span_id() — returns the open span's TraceContext (synthetic ids) or None - add_event_to_current_span(name, **fields) — drops an event on the open span

ScriptExhausted

Bases: BaseException

A FakeLLM.script was asked for more turns than it has.

WHY THIS EXISTS, AND WHY IT IS LOUD

This class used to be a min(). stream clamped the script index, so a 2-turn script asked for 4 turns answered ['one', 'two', 'two', 'two'] — measured, not hypothetical. A scripted LLM exists to drive an agent through an EXPECTED number of turns, so "the agent asked for more turns than the script has" is a statement about the code under test: it looped when it should have stopped. The clamp turned exactly that finding into a stable, plausible, wrong answer, and the test went green. The harness built to catch non-termination was the thing hiding it — the same shape as a broad except that reports success.

WHY BaseException AND NOT Exception

Because the failure it reports is a failure of the code that would be doing the catching. agentkit has ~20 deliberate except Exception sites on the path between FakeLLM.stream and the test body — react reflects invalid output back to the model, _invoke_tool_safe turns every tool failure into a tool message, resilience classifies pre-stream failures into retry-vs-fail. Each one is correct for a real provider fault, and each one would convert "your loop does not terminate" into a retry or a reflected message and keep going. asyncio.CancelledError was promoted out of Exception in 3.8 for this exact reason; a test double's contract violation belongs in the same bucket. pytest reports a BaseException as a plain failure, so nothing is lost at the top.

Turn dataclass 

Turn(
    content: str = "",
    tool_calls: tuple[ToolCall, ...] = (),
    usage: Usage | None = None,
)

One scripted reply for FakeLLM.script — a final answer, or a turn that requests tools.

ScriptedAdapter 

ScriptedAdapter(turns: Iterable[Turn])

A finite sequence of transport facts; exhaustion fails loudly.

ScriptedAdapterExhausted

Bases: BaseException

A runtime requested more transport turns than the script supplied.

codex_turn 

codex_turn(
    *,
    text: str | None = None,
    reasoning: str | None = None,
    thread_id: str | None = DEFAULT_THREAD_ID,
    usage: tuple[int, int, int] | None = None,
    items: Iterable[Mapping[str, Any]] = (),
    failed: str | None = None,
    error: str | None = None,
    duration_ms: int | None = None,
) -> tuple[dict[str, Any], ...]

One complete, well-formed codex exec --json turn, as payloads.

Emits the real event order: thread.started (unless thread_id=None, which is how a test builds the resumed turn of a CLI version that does not re-announce the thread), turn.started, one item.completed per item, then exactly one terminal turn.completed or turn.failed.

usage is (input_tokens, cached_input_tokens, output_tokens) in Codex's own convention — input INCLUSIVE of the cached prefix. Passing it that way rather than as agentkit's split is deliberate: the subtraction is the cognition's job, and a builder that pre-split it would make the one test that exists to pin that subtraction assert against its own arithmetic.

reasoning and text are shorthand for the two items nearly every test wants; items takes the rest (a command_execution, a file_change, an mcp_tool_call) as raw item dicts, appended after them. error inserts a top-level stream error before the terminal event — the shape the CLI uses for a broken pipe, as opposed to failed, which is the turn itself going wrong.

make_test_ctx 

make_test_ctx(
    *,
    llm: Any = None,
    invoker: Any = None,
    scope: Scope | None = None,
    budget: Budget | None = None,
    store: Any = None,
    vector: Any = None,
    checkpointer: Any = None,
    asker: Any = None,
    observer: Any = None,
    trace: Any = None,
    cancel: CancellationToken | None = None,
    chat_middleware: Sequence[Any] = (),
    tool_middleware: Sequence[Any] = (),
    meters: Sequence[Any] = (),
    correlation_id: str = "test-run",
    autonomy: AutonomyLiteral = "auto",
) -> RunContext

Build a RunContext for tests with the knobs the call site cares about and sensible no-op defaults for everything else.

Two ways to wire the LLM seam: pass invoker= (a fully-built Invoker) OR pass llm= and we'll wrap it with Invoker(llm=..., chat_middleware=chat_middleware). Tests that don't need an Invoker at all leave both unset — they get a RunContext with services.invoker = None, which is fine for capabilities tests (RequestBuilder, Compactor) that never invoke an LLM.

Scope() is the zero-tenant default. Override with scope=Scope(org_id, domain_id) for tenant-isolation tests.

asker wires the human-in-the-loop transport (agents.control.elicitation.Asker). When set, a gated tool call PARKS — the cognition awaits it in place instead of checkpointing and unwinding. None (default) keeps the classic suspend-and-resume path.

observer / trace / checkpointer / store / vector default to None here, which we omit when constructing Services so its built-in NoopObserver / NoopTrace factories still kick in — passing None explicitly would clobber them and break code that calls ctx.trace.span(...) or ctx.observer.emit(...).

scripted_runtime 

scripted_runtime(
    turns: Iterable[Turn],
    pricing: PriceTable,
    *,
    mode: Mode = OPENAI_API,
    **runtime_options: Any,
) -> CallRuntime

Create a runtime wired to one scripted mode and an explicit price table.