> ## Documentation Index
> Fetch the complete documentation index at: https://docs.plurai.ai/llms.txt
> Use this file to discover all available pages before exploring further.

# Write a harness

> The simulate.py contract: the hooks the runner calls, what each receives and returns, and the plurai helpers a hand-written harness uses.

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.

> [!TIP]
>
> 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:

**1. Stub production dependencies**

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

```python theme={null}
import sys
from unittest.mock import MagicMock

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

**2. 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.

```python theme={null}
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.

**3. 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](https://docs.plurai.ai/integrations/agent-frameworks.md).

**4. 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

```python theme={null}
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.

> [!WARNING]
>
> 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

```python theme={null}
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`.

```python theme={null}
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:

```bash theme={null}
plurai-run-simulation --scenario-set-id <id> --scenario-id <id> \
  --agent-module-path plurai_sim/simulate.py --out /tmp/plurai-check
```

## Next steps

> [card] [API reference](https://docs.plurai.ai/sdk/api-reference.md)
>
> Every supported name in the `plurai` package.

> [card] [Configuration](https://docs.plurai.ai/reference/config.md)
>
> `plurai.config.yaml`, including `world_state` and `agent.io`.


This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.