Testing and Replay Tests

Three days in a second

The agent runtime parks at a checkpoint and waits up to 72 hours for a human to approve the plan. You cannot test that by waiting. You also cannot test the agent's decisions by paying for an LLM call on every CI run, and you cannot find out whether last month's histories still replay under today's code by deploying and watching.

Upstream sorts automated tests into end-to-end (a real Server and Worker, driven from a Client), integration (anything between: Workers with mock Activities, Activities with a mocked context) and unit (one function, everything it calls mocked), and recommends writing the majority as integration tests. This course cuts the same ground by what you are testing: Workflow tests, Activity tests, and replay tests. The third kind is the one most teams have never written, and it is what stands between you and a broken deploy.

Workflow tests: time skipping and mocked Activities

WorkflowEnvironment.start_time_skipping() starts a test server process that is an in-memory implementation of Temporal Server with time skipping. In time-skipping mode, Timers, which include sleeps and conditional timeouts, are fast-forwarded except when Activities are running. A Workflow that sleeps for 72 hours completes in milliseconds, and its history still shows a 72-hour timer.

Mock the Activities by registering functions with the same name and signature as the real ones. The Workflow invokes the mock instead of the real implementation, so you can test a decision: given these tool results, the agent takes this branch.

import uuid

import pytest
from temporalio import activity
from temporalio.testing import WorkflowEnvironment
from temporalio.worker import Worker

from agent_runtime.workflows import AgentRun, ToolCall, ToolResult

calls = {"execute_tool": 0}


# Same name and signature as the real call_llm; no model is called.
@activity.defn(name="call_llm")
async def call_llm_mocked(prompt: str, key: str) -> str:
    if "RESULT:" in prompt:
        return "FINAL: the saga is compensation in reverse"
    return "TOOL: search('temporal saga pattern')"


@activity.defn(name="execute_tool")
async def execute_tool_mocked(call: ToolCall) -> ToolResult:
    calls["execute_tool"] += 1
    return ToolResult(key=call.key, output="RESULT: 3 documents")


@pytest.mark.asyncio
async def test_agent_stops_after_one_tool_call():
    task_queue = str(uuid.uuid4())
    async with await WorkflowEnvironment.start_time_skipping() as env:
        async with Worker(
            env.client,
            task_queue=task_queue,
            workflows=[AgentRun],
            activities=[call_llm_mocked, execute_tool_mocked],
        ):
            result = await env.client.execute_workflow(
                AgentRun.run,
                "explain sagas",
                id=str(uuid.uuid4()),
                task_queue=task_queue,
            )
            assert calls["execute_tool"] == 1
            assert result.startswith("the saga")


@pytest.mark.asyncio
async def test_checkpoint_expires_after_72_hours():
    task_queue = str(uuid.uuid4())
    async with await WorkflowEnvironment.start_time_skipping() as env:
        async with Worker(env.client, task_queue=task_queue,
                          workflows=[AgentRun],
                          activities=[call_llm_mocked, execute_tool_mocked]):
            handle = await env.client.start_workflow(
                AgentRun.run, "needs approval",
                id=str(uuid.uuid4()), task_queue=task_queue,
            )
            # Never send the approve signal: the 72 h timer fires at once.
            result = await handle.result()
            assert result == "expired: no approval within 72h"

Two upstream cautions. Time is a global property of an instance of WorkflowEnvironment: skipping time applies to every test running against that instance, so run tests that need manual time control (await env.sleep(...)) in series or on separate instances. And start_time_skipping() is not the only option: start_local() runs a full local Server with real time (pass ui=True to get the Web UI), and from_client() wraps an existing Server.

Activity tests: ActivityEnvironment

An Activity can be tested with a mock Activity environment, which provides a way to mock the Activity context, listen to Heartbeats, and cancel the Activity. No Worker, no Server; you call the function through ActivityEnvironment().run(...) and it sees a fake context.

Set on_heartbeat to capture every activity.heartbeat() call, and call env.cancel() to deliver a cancellation the way a heartbeat response would in production. This is the test for module 7's long-running tool:

import asyncio

import pytest
from temporalio.testing import ActivityEnvironment

from agent_runtime.activities import release_gpu, run_long_tool


@pytest.mark.asyncio
async def test_long_tool_heartbeats_and_stops_on_cancel():
    env = ActivityEnvironment()
    heartbeats = []
    env.on_heartbeat = lambda *args: heartbeats.append(args[0])

    task = asyncio.create_task(env.run(run_long_tool, "fine-tune adapter"))
    await asyncio.sleep(0.2)
    env.cancel()                      # what the Service would do on the next heartbeat
    with pytest.raises(asyncio.CancelledError):
        await task
    assert heartbeats, "a long tool must heartbeat or it can never be cancelled"


@pytest.mark.asyncio
async def test_release_gpu_is_idempotent(fake_scheduler):
    env = ActivityEnvironment()       # one environment per test: cancel() is sticky
    await env.run(release_gpu, "gpu-reservation-41")
    await env.run(release_gpu, "gpu-reservation-41")
    assert fake_scheduler.release_calls == 1

The first test checks the two facts module 7 relied on: the Activity heartbeats, and it re-raises on cancel. The second is the at-least-once test from module 4 in its cheapest form: run it twice, assert one effect.

Replay tests: histories against new code

Replay recreates the exact state of a Workflow Execution from the beginning of its Event History. Replay succeeds only if the Workflow Definition is compatible with the provided history from a deterministic point of view. That makes it the deploy gate: a history that replays clean under the new code can be picked up by a Worker running that code; one that fails needs a patch (module 9).

First export a history. With the Temporal CLI, JSON output is explicitly meant to be passed to an SDK for replay:

temporal workflow show \
    --workflow-id agent-42 \
    --output json > tests/histories/agent-42.json

The Web UI's download button produces the same JSON. Then replay it with Replayer, passing the file as a JSON string or as the dict from json.load():

import json
from pathlib import Path

import pytest
from temporalio.client import WorkflowHistory
from temporalio.worker import Replayer

from agent_runtime.workflows import AgentRun

HISTORIES = Path(__file__).parent / "histories"


@pytest.mark.asyncio
@pytest.mark.parametrize("path", sorted(HISTORIES.glob("*.json")), ids=lambda p: p.stem)
async def test_history_replays_under_current_code(path: Path):
    history = WorkflowHistory.from_json(path.stem, json.loads(path.read_text()))
    replayer = Replayer(workflows=[AgentRun])
    await replayer.replay_workflow(history)   # raises on non-determinism

For a live check against the Server instead of fixtures, list executions and replay them all. Upstream's recommended CI recipe is exactly this: decide which Workflow Types or Task Queues the Worker code under test targets, download the Event Histories of a representative set of recent open and closed Workflows from each, run them through replay, and fail CI if any error is encountered.

workflows = client.list_workflows(
    "TaskQueue='agent-runs' and StartTime > '2026-08-01T00:00:00'"
)
results = await Replayer(workflows=[AgentRun]).replay_workflows(
    workflows.map_histories(), raise_on_replay_failure=False
)
assert not results.replay_failures, results.replay_failures

The failure you are looking for reads like the one the upstream sample provokes by adding a sleep before an Activity: NondeterminismError('... Timer machine does not handle this event: HistoryEvent(id: 8, Some(ActivityTaskScheduled))'). Lab 8.2 makes you produce that error on purpose, from a history you recorded under v1, before you ever deploy.

A histories fixture is an asset

Check exported histories into the repo under tests/histories/, one per interesting path: a clean run, a run with a tool retry, a run that hit the 72-hour checkpoint, a run that was cancelled mid-tool. Every Workflow change runs the replay test above against all of them in CI.

The fixture grows when an incident produces a history worth keeping. A production execution that failed a Workflow Task is a free regression test once you export it. Keep the Workflow ID in the filename so from_json gets it for nothing.

Forensic debugging from history

The replay test tells you whether a history can be continued. Sometimes you need to read one when nobody can hand you the code: the Worker image that produced it was garbage-collected, the team that owned it is gone, or you are on call for a service you have never opened. Lab 8.5 gives you a broken AgentRun history — a JSON export, no source — and five questions. They are the same five module 2 promised you could answer from the log alone, and each has a place in the history where its answer sits.

1. Which attempt failed? Find the last ActivityTaskScheduled without a terminal event, then temporal workflow describe for its pending-activity record: attempt count, last failure message, last heartbeat time. The attempt number is not in the history; it is mutable state the Service keeps beside it.
2. Was it retried? The retry policy is in the ActivityTaskScheduled attributes (--detailed). Attempts beyond one leave no Events of their own, so "retried three times" is read from the attempt counter in describe, not from the log.
3. Did the side effect possibly occur? Every ActivityTaskStarted is an attempt that ran. For the pending one, the Service cannot tell you: the Worker may have charged, embedded, or written before it died. Read the heartbeat details for how far it got, then the idempotency key in the input to say whether a retry would double it. The definitive answer is in your system of record, not in Temporal.
4. What is the state now? Replay it in your head: walk the Events, apply each Completed result to the Workflow's variables, apply each WorkflowExecutionSignaled in arrival order. temporal workflow stack gives the line the code is blocked on if a Worker is alive; the Search Attributes give the last published AgentStatus and CurrentStep even if none is.
5. What if a Worker starts now? It replays 1–N, reaches the first Command with no matching Event, and issues it. Predict that Command from the history; then say whether it is safe — a retry of a tool with a key, or a fresh charge without one.

Write the five answers down before you look at the source. Then diff your answers against a replay test with the source restored: the two places you were wrong are the two things the history does not record, and those are the things your own Activities must record instead.

Failure injection checklist

Upstream's pre-production guidance is built on one idea: the goal is not just to verify that things fail and recover, but to build confidence that recovery, correctness, consistency and operability hold under real-world conditions. Partial failure is often harder than total failure, and recovery paths deserve as much testing as steady state. Build observability before you break things.

Each of these is a test in Lab 8.4, not a war story:

• Worker crash. Kill every Worker mid-Activity, then restart them. Watch for duplicate or improper Activity results, Workflow failures, and backlog growth and drain time. This validates at-least-once semantics and that Workflows replay cleanly.
• Frequent Worker restart. Restart a fraction of the fleet every few minutes. Watch replay latency; this is what forces the sticky queue to be invalidated and tasks rescheduled.
• Activity timeout. Make a tool exceed its start_to_close_timeout. Watch retry and backoff behaviour, and check that API timeouts and Activity timeouts are set against each other deliberately.
• Network error / broken dependency. Make the model endpoint return 5xx or hang. Temporal guarantees Workflow durability, not dependency availability; this is where you find infinite retries without circuit breaking and Workflow logic that "waits out" a broken dependency.
• Duplicate execution. Run the same tool call twice with the same key. The happy path never touches a non-idempotent tool; this injection does.
• Cancellation. Cancel during a long tool call, with and without heartbeats, and assert the cleanup Activity ran.
• A version that causes non-determinism. Deploy Workflow code that would introduce non-deterministic errors, then recover with rollback or a patch. Module 9 is this bullet.

Introduce one variable at a time, record start and stop times, and afterwards update the retry, timeout or versioning policy the experiment exposed.

What the lab will test

Name the cheapest test that can catch each bug: wrong branch after a code change, lost heartbeat progress, duplicate external effect. In Lab 8, make each bug visible with the appropriate test.

Try it

Lab 8 lives in labs/08-testing-and-replay-tests/ and has five exercises:

• 8.1 — time-skipping Workflow tests for the agent loop with mocked Activities, including the 72-hour checkpoint finishing in under a second.
• 8.2 — export a history under v1, change the Workflow, watch the replay test fail with the same error a Worker would raise; then patch it (a preview of module 9).
• 8.3 — Activity tests with ActivityEnvironment: captured heartbeats, delivered cancellation, and an idempotency check.
• 8.4 — the failure-injection suite: crash, timeout, duplicate, cancel.
• 8.5 — forensic: a broken AgentRun history, no source; answer the five questions in writing from show, describe and stack alone, then check them against a replay with the source.

Lab page: Open the guided lab — the exercises, every source file, and the local setup steps.