# Experiment Framework

`plesty.lib.experiment` turns a measurement routine into a **reproducible,
crash-tolerant run**. The mental model is checkpointed training in deep
learning: the *plan* is the training schedule, the *journal* is the training
log, and the completed-step set plus persisted step results form the
checkpoint.

An `Experiment` orchestrates devices through a
[`CompositeDevice`](device/composite_device.md) — it receives Device
instances, never raw hardware.

## The pieces

| Piece | Role |
|---|---|
| `Step` | The smallest unit of work: a stable `id`, an operation name `op`, and its `params`. A step either completes (result persisted and journaled) or is re-run on resume. |
| `Plan` | The frozen, ordered schedule of steps plus the run configuration. Written to disk at run start, never mutated; a SHA-256 `content_hash()` identifies the schedule. |
| `Journal` | A crash-safe, append-only JSONL event log (`run_started`, `step_started`, `step_completed`, `step_failed`, `run_completed`/`_canceled`/`_aborted`). Replaying it derives the run state and the completed-step set. |
| `records.jsonl` | One line per completed step: its result document (`plesty.lib.data.append_record` / `read_records`). One file, appended and fsynced per step, tailed by offset — not one small file per step, which is slow to write and slow to follow on a network share. |
| `Experiment` | The async ABC tying it together: validate → journal → execute with retries → persist results → resume. |
| `Run` | Read access to a stored run — plan, config, journal status, records, and the translation of device-written paths onto this machine (`plesty.lib.experiment.runs`, which also spells out the naming below and finds runs: `list_runs`, `latest_run`). |

Every run owns a directory:

```text
<run_root>/<run_id>/            # run_id = <name>_<YYYYmmdd-HHMMSS>
├── plan.json                   # frozen schedule + config + content hash
├── journal.jsonl               # append-only event log
├── records.jsonl               # one line per completed step: its result document
└── data/                       # blobs the records reference (arrays, images), if any
    └── step_0003.npy
```

Each record line is a typed [`ResultDocument`](data_schemas.md) with
provenance (`step_id`, `op`, `params`) plus the step `index`; a JSON value
result is inlined, an array or image stays a raw blob in `data/` that the
line references. The line is the commit — it is appended (and fsynced) only
after the blob is on disk, and a resumed run treats a committed record as a
completed step even if the crash came before its journal event. Runs
written by plesty-lib ≤ 0.3.4 (one `data/step_*.json` per step) are still
read by `RunSource`, `records()`, and `Run`.

## Writing an experiment

Subclasses override exactly three lifecycle hooks — `build_plan()`
(required), `setup()`/`teardown()` (optional, extend the defaults which
connect/disconnect all devices) — plus one public method per step operation:

```python
import asyncio
from plesty.lib.data import PlestyArray
from plesty.lib.experiment import Experiment, Plan, Step

class LineScan(Experiment):
    def build_plan(self) -> Plan:
        steps = [
            Step(id=f"scan[x={x}]", op="scan_point", params={"x": x})
            for x in range(5)
        ]
        return Plan(steps, config={"points": 5})

    def scan_point(self, x: int) -> PlestyArray:
        # drive devices via self.devices, return the measurement
        return PlestyArray([float(x)], name="signal", unit="a.u.")

run_id = asyncio.run(LineScan(run_root="runs").run())
```

Rules that keep runs reproducible:

- **`build_plan()` must be deterministic** for a given configuration —
  resume compares the regenerated plan against the persisted one by content
  hash.
- **Step ids must be stable** (e.g. `"scan[x=3,y=5]"`) — never derived from
  run time or randomness; resume matches completed steps by id.
- **Step operations are validated before anything runs**: every `Step.op`
  must name an existing public method and may not be a reserved lifecycle
  method (`build_plan`, `run`, `setup`, `teardown`). Violations raise
  `InvalidOperationError` listing all offending steps — before any device is
  touched or journal written.

Step methods may be sync or async; results are awaited automatically. Each
step is retried (`max_retries`, `retry_sleep` constructor arguments) before
the run aborts.

## Starting on a rig that is really there

An experiment's `__main__` builds the rig with `connect_rig` — the same
start for every experiment: connect the servers, run the composite's
`preflight()` (each sub-device exposes the methods its configuration
`requires` and answers `identity()`), and if anything is missing, tell the
operator which device, at which address, from which variable — and ask:

```python
from plesty.lib.experiment import connect_rig

rig, skipped = connect_rig(
    lambda skip: PolPlRig(with_powermeter="pm" not in skip),
    optional={"pm"},                       # the experiment can run without it
)
```

```text
The rig is not ready — 1 device(s) failed the preflight:
  - pm: no answer from tcp://127.0.0.1:5557: TimeoutError('Server did not respond in time')
      address tcp://127.0.0.1:5557 (from PM_ADDRESS in .env / the environment)
Check that the pm server(s) are running and that PM_ADDRESS point at them
(a wrong port connects fine but is not the expected device).
[r] retry (after fixing .env or starting the server)  [a] abort  [s] continue without pm
>
```

`r` rebuilds the rig (an edited `.env` or a freshly started server is picked
up), `a` raises `RigNotReadyError`, `s` — offered only when every failing
sub-device is in `optional` — returns the rig without them and names them in
`skipped`, so the experiment can drop the matching part of its plan. Without
a terminal on stdin nothing waits: the diagnosis is logged and
`RigNotReadyError` raised.

The check also runs inside `Experiment.setup()`: a run never starts on a
composite whose `preflight()` reports a problem, whether or not `connect_rig`
was used. Declare what each sub-device must be able to do in its
configuration — that is what turns "the port answers" into "it is the
stage":

```yaml
devices:
  hwp:  {timeout_ms: 5000, requires: [home_stage, move_absolute, get_position]}
  spec: {timeout_ms: 5000, requires: [acquire, get_recent_file, write, query]}
  pm:   {timeout_ms: 5000, requires: [measure_power, write]}
```

## Data on a shared disk

Blobs a device writes itself (spectrometer frames, camera images) never
cross the network: the acquiring host saves them to a shared disk and the
step record carries only the path. Two machines spell that share
differently, so it is named once, by environment:

| Variable | Meaning |
|---|---|
| `PLESTY_DATA_DIR` | the share as the acquiring host sees it (e.g. `G:\RAWDATA`) |
| `PLESTY_DATA_MOUNT` | the same share as this machine sees it (e.g. `/mnt/group/RAWDATA`) |

The experiment reads both (or takes `data_dir=` / `data_mount=` at
construction). With the mount set, `run_root` defaults to it, so runs land
on the share as well. Before `setup()` is awaited the run id is known and
`self.raw_dir` names `<data_dir>/<run_id>/raw` in the host's spelling — one
folder per run, the same one on resume — which `setup()` routes the writing
devices to:

```python
async def setup(self) -> None:
    await super().setup()
    if self.raw_dir is not None:
        self.devices.set_data_path(self.raw_dir, devices=["spec"])
```

`run_started` journals `data_dir`, `data_mount`, and `raw_dir`, so a reader
needs at most its own mount:

```python
from plesty.lib.experiment import Run, latest_run

run = Run(latest_run("/mnt/group/RAWDATA", name="pol_pl"))
run.config["powermeter"]                 # the frozen configuration
run.status()["state"]                    # "running", "completed", ...
for record in run.records():
    run.local(record["data_file"])       # G:\RAWDATA\... → /mnt/group/RAWDATA/...
```

`Run.local` uses, in order, an explicit `mount=`, `PLESTY_DATA_MOUNT`, and
the mount journaled by the machine that ran the experiment (only when that
directory exists here); with neither a data root nor a mount the path is
returned as recorded — the case of reading on the acquiring host itself.
`run.source(mapper)` returns a `RunSource` (live) or, with
`Run(..., replay_rows=n)`, a `ReplaySource` — see [Monitors](monitor.md).

## Crash, cancel, resume

`run()` returns the run id. If the run is interrupted — `Ctrl-C`, a crash,
an aborting step — pass the id back to continue:

```python
run_id = asyncio.run(LineScan(run_root="runs").run())
# ... interrupted?
asyncio.run(LineScan(run_root="runs").run(resume=run_id))
```

On resume the framework:

1. regenerates the plan and **refuses to continue if its content hash
   differs** from the stored one (`PlanMismatchError`) — a silently changed
   schedule or configuration cannot corrupt a run;
2. replays the journal to find completed steps and skips them;
3. journals the resume (`run_started` with `resumed: true`) and continues.

`teardown()` is always awaited — on completion, failure, and cancellation —
so devices are released no matter how the run ends.

## Contract tests without hardware

`plesty.lib.test.experiment_pipeline.ExperimentPipeline` verifies the
plan/checkpoint contract of an experiment module with **no instrument
attached** (it never calls `run()`). Five gates:

| Gate | Checks |
|---|---|
| 1 `test_experiment_subclass` | public `Experiment` subclass, instantiable without devices |
| 2 `test_plan_deterministic` | `build_plan()` twice → identical `content_hash()` |
| 3 `test_plan_ops_resolve` | every `Step.op` resolves via the framework's own plan validation |
| 4 `test_plan_serializable` | `Plan.save`/`load` round-trip preserves the hash |
| 5 `test_lifecycle_hooks` | `setup`/`teardown` overrides are async |

Experiment HUB modules expose one pytest function per gate (SDK gate E1
verifies their presence and passage — the experiment-tier mirror of device
gate d1); see [Test Helpers](device/test_helper.md) for the pattern.