Write a harness

The harness, simulate.py, is the one file between the runner and your agent. The scan generates it for you; this page is the contract for when you write one by hand or edit the generated one.

You rarely need to start from scratch. Run /simulation-scan-agent and edit what it writes: it also checks the harness against this contract with its verifiers and a smoke run.

How the runner uses it

simulate.py is not a script you launch. The runner starts it in a child process (python -m plurai.child simulate.py), imports it, and calls its hooks over a private channel on the child’s standard input and output. So:

Do not print to stdout from the harness or anything it spawns. stdout is the runner’s channel. Send logs to stderr or to a file you control.
No if __name__ == "__main__": block is needed.
Never call asyncio.run() inside a hook. The runner already drives an event loop. Any hook may be async def; the runner awaits it.
Load your agent’s .env yourself. The child imports only simulate.py, so settings that live only in .env (model, provider keys) are invisible unless the harness loads them, for example with dotenv.load_dotenv().

Structure

A harness does its work in this order, at import time:

Stub production dependencies

Replace modules that connect to production on import with MagicMock, before anything imports them:

import sys
from unittest.mock import MagicMock

sys.modules["our_backend.db"] = MagicMock()

Define the mocks

Each mocked tool is a small @plurai.mock handler that reads the scenario’s world. Keep handlers simple; the rich description of the tool (what it reads and writes, its business rules) lives in its descriptor.

import plurai

@plurai.mock("get_user_profile", descriptor={
    "description": "Fetch a user profile",
    "args": {"user_id": {"type": "str"}},
    "returns": {"type": "dict"},
})
def mock_get_user_profile(user_id: str):
    ctx = plurai.get_simulation_context()
    return ctx.world_state.get_entity(user_id) or {"error": "not found"}

A handler reads the world three ways: ctx.world_state.get_entity(id), ctx.world_state.query(type, **filters), or plurai.world_state_snapshot() for the whole map. It writes with ctx.world_state.apply({id: {...fields}}).@plurai.observe("rag_search") registers an observer instead: the real tool runs, and the observer records its result into the world.

Route your agent's tools to the mocks

Patch tool functions before your agent is imported (plurai.apply_mocks(module, {"name": "mock", ...})), or swap a LangGraph ToolNode for plurai.MockToolNode. See Python agent frameworks.

Import your agent and export the hooks

Import your agent unchanged, then define invoke_agent and any optional hooks below.

Hooks

The runner discovers these functions by name on the module.

Hook	Required	Called
`invoke_agent(payload, config=None)`	One of `invoke_agent` or `invoke_pipeline`	Once per conversation turn.
`invoke_pipeline(inputs)`	One of `invoke_agent` or `invoke_pipeline`	Once per scenario, for a one-shot pipeline.
`seed_state(initial_agent_context, initial_world_state, resume_seed=None)`	When your harness holds state the mock does not, and for resumed scenarios	Once per scenario, before the first turn.
`reset()`	When `plurai.config.yaml` declares `world_state.owner: agent` or `external`	Once per scenario, before `seed_state`.
`world_snapshot()`	Same as `reset()`	Once per scenario, at the end.

invoke_agent

def invoke_agent(payload, config=None):
    return app.invoke(payload, config={**(config or {})})

Payload

payload["messages"] holds only the new user message, never the conversation so far. It is empty on the opening call when the agent speaks first (agent.io: {may_open: true}), so never index [-1] blindly. config["configurable"]["thread_id"] is stable for the whole scenario. The scenario’s agent context is spread into the payload on turn 1 only.

History is your harness’s job. Keep it keyed on thread_id: a LangGraph checkpointer such as MemorySaver does this natively; a framework-free harness keeps its own store. A harness that rebuilds history from payload["messages"] runs stateless, and nothing flags it: the Run completes and every memory or continuity verdict is meaningless.

Other payload keys:

Key	When
`interrupt_choice`	The turn answers an interrupt your agent raised. It holds one of the choices you offered, verbatim. Map it to your resume value; never parse the user’s words.
`channel_inputs`	Values planted through input channels declared under `agent.io.channels`.

A document the user hands over arrives in the new message’s content as a file part with a data: URI. It is never an agent-context key or a seed_state value. Declare it with agent.io: {input: [text, file]} in plurai.config.yaml.

Return value

Return your agent’s own output, unaltered:

the agent state dict with the message list under messages (LangGraph’s result), or
a plain string, or
a message list under messages in your framework’s own shape: LangChain objects, OpenAI role dicts or Anthropic content blocks. The SDK converts them; do not normalize.

A turn where your agent only acted (a render, a hand-off) and said nothing is complete: return content: "" with its tool calls. Never paint actions into the reply text, and never invent a sentence for a silent turn. Every key of a returned dict other than messages is saved per turn in transcript.json as agent_state. These keys have a defined meaning:

Key	Meaning
`tool_trail`	For an agent that runs its own tool loop and returns prose: one entry per dispatch, in order, `{"tool", "args", "call_id", "result", "status"}`. `call_id` is required and unique per dispatch. `status` is `ok`, `error`, `not_dispatchable` (a name your gate refused because the agent does not have that tool) or `denied` (your approval gate said no; add `"refusal": {"by": "harness", "reason": ...}`).
`input_context`	Content your agent’s own code added to its model input this turn (a memory, a state snapshot, pre-retrieved documents), as `[{"source": ..., "content": ...}]`. `plurai.model_input_context` builds it.
`appended_output`	Text your product’s code appended to the reply after the model, as `[{"source": "product", "site": "<file:line>", "text": ...}]`. Keep the text in `messages` too.

For trace platforms to show each model call, return the turn’s whole chain: each assistant message with its tool calls and the model and token usage your framework reports. A LangGraph or LangChain result already carries them. Sync or async: write def invoke_agent for graph.invoke, async def invoke_agent for graph.ainvoke. Pass your production entry’s run settings, step cap included; never raise a cap such as LangGraph’s recursion_limit for the simulation.

invoke_pipeline

invoke_pipeline(inputs) -> dict, sync or async, for a one-shot agent with no conversation: the runner calls it once with the scenario’s inputs and scores the (inputs, outputs) pair. Declare shape: pipeline in plurai.config.yaml. A bench that exports invoke_pipeline and no invoke_agent without declaring a shape is refused at Run start.

seed_state

def seed_state(initial_agent_context, initial_world_state, resume_seed=None):
    entities = (initial_world_state or {}).get("entities") or {}
    if entities:
        plurai.world_state.apply(entities)
    if not resume_seed:
        return None
    ...  # replay resume_seed into your agent's history for this thread
    return {"resume_state_dropped": dropped} if dropped else None

Three positional parameters, always. The third is passed on every scenario, None for a fresh one. A two-parameter hook fails every Run during setup, and the SDK names the required signature.
initial_agent_context holds the scenario’s context values, including those declared inject: request.
initial_world_state["entities"] is the scenario’s seeded world. Apply it with plurai.world_state.apply(...), the same surface your mocks read. The world is per scenario, so there is nothing to clear first.
resume_seed is the conversation a resumed scenario starts from (a scenario built from a production trace). Apply it to your agent’s history; what “apply” means is your harness’s decision. Return None when it was restored in full, or a dict with string keys saying what was left out. Raise plurai.HarnessFault when fewer messages are restored than seeded.
A harness with no seed_state runs fresh scenarios and is refused on a resumed one.

When your own store needs a seeded row as a record, convert it with plurai.store_record(row), which drops the routing keys (_entity_type and the backend’s id copy). Keep the original row for routing and for world_snapshot().

reset

reset() takes no arguments and returns nothing. It clears what your harness owns before each scenario: a file, a database table, a checkpointer’s database, a store on a real service. Never clear plurai.world_state here; the backend re-seeds it per scenario. A config that declares world_state.owner: agent or external with no reset() is refused at Run start. Up to concurrency × trials copies of your harness run at once. Anything the harness creates (a port, a working folder, a lock file, a record) must be per child: bind port 0, use tempfile.mkdtemp(), and tag records with plurai.active_session_id().

world_snapshot

world_snapshot() takes no arguments and returns the world your harness owns as {entity_id: {...fields}}, each value carrying _entity_type. The runner calls it once at the end and reports it as the scenario’s final world, which world-reading evals are scored against. It must be JSON-serializable. Without it, on a bench that declares its own world, world-reading evals resolve to not-measured instead of scoring an empty world. For a file your agent wrote, add one row with plurai.file_row(path, entity_type="output_file"). For a file handed to your agent, pass preview=False. Declare the row’s fields in world_state.entity_types.

def world_snapshot():
    world = {f"ticket_{row['id']}": {"_entity_type": "ticket", **row} for row in store.all()}
    for path in OUTPUT_DIR.rglob("*"):
        if path.is_file():
            rel = path.relative_to(OUTPUT_DIR).as_posix()
            world[f"output_file:{rel}"] = plurai.file_row(path, entity_type="output_file")
    return world

If part of the world lives in the mock, declare the types your store holds under world_state.owned_types and return only your own rows; the Run combines them with the mock’s rows of every other type.

Faults: whose failure is it

The Run separates failures that belong to your agent from failures that belong to the harness.

Raise	When	Recorded as
`plurai.HarnessFault("...")`	The harness itself refuses: a path the bench does not cover, a value it cannot render, an unmocked body reached.	`not_valid`. The scenario is not scored against your agent. It is recorded when constructed, so an agent tool that catches it does not hide it.
`plurai.AgentFault("...")`	Your agent’s own code failed where no Python frame shows it, for example in a bridge child in another runtime, and the harness attributes it to the agent. Raise it from `invoke_agent`.	`agent_raised`: a failed trial your agent owns. A provider failure, or a `HarnessFault` recorded earlier in the turn, still wins.
Anything else from a hook		Your agent’s failure.

A mock that refuses a world write is recorded the same way as a HarnessFault.

Mock results and errors

plurai.mock_result(response) turns a mock response into what your agent sees: the tool’s value, or a raised plurai.MockToolError for a failed call. How a failure reaches your agent follows the tool’s declared error_channel in plurai.config.yaml: raises raises MockToolError as the real tool would; returns hands back the failure value. Catch MockToolError in the harness and give your agent its own failure, never a value that reads as success.

Check your harness

The plugin’s scan runs two verifiers and a smoke run over a generated or edited harness. To run them after a manual edit, ask in chat: “Verify my simulate.py.” Then run one scenario:

plurai-run-simulation --scenario-set-id <id> --scenario-id <id> \
  --agent-module-path plurai_sim/simulate.py --out /tmp/plurai-check

Next steps

API reference

Every supported name in the plurai package.

Configuration

plurai.config.yaml, including world_state and agent.io.

Use the SDK

Reference

How the runner uses it

Structure

Hooks

invoke_agent

Payload

Return value

invoke_pipeline

seed_state

reset

world_snapshot

Faults: whose failure is it

Mock results and errors

Check your harness

Next steps

API reference

Configuration

​How the runner uses it

​Structure

​Hooks

​invoke_agent

​Payload

​Return value

​invoke_pipeline

​seed_state

​reset

​world_snapshot

​Faults: whose failure is it

​Mock results and errors

​Check your harness

​Next steps

API reference

Configuration

How the runner uses it

Structure

Hooks

invoke_agent

Payload

Return value

invoke_pipeline

seed_state

reset

world_snapshot

Faults: whose failure is it

Mock results and errors

Check your harness

Next steps