plesty.server.services.experiments

Running experiments on this bench, and reading what they are doing.

An experiment run is a job like any other — a detached process whose output lands in logs/jobs/<id>.log — so nothing here re-implements launching or log tailing. What is particular to an experiment is on either side of that:

Before, the module is asked what it takes. python -m <module> --manifest runs in the module’s own environment and prints the declaration; the console builds its settings from that and holds no per-experiment code. The answer is read fresh rather than cached in the fleet, so an updated module cannot leave a stale copy of its own contract behind.
After, the run is watched and driven through its run directory. journal.jsonl, records.jsonl and progress.json say what it is doing, and control.json says what it should do next. Both are files in a directory this bench owns: one read per refresh, nothing on the wire, and a console that crashes leaves a run that carries on.

Stopping is not cancelling the job. JobRunner.cancel kills a process tree, which for a measurement means no run_canceled in the journal and so no point to resume from. A run is stopped by asking it to stop — which it notices at a step boundary, where the journal has just committed — and the kill stays for a process that has stopped answering.

Attributes

`MANIFEST_TIMEOUT`
`JOB_KIND`
`VIEWER_KIND`
`CONTROL_FILE`
`CANCEL`
`ABORT`
`CREATED_FILE`
`RUN_CONFIG_FILE`
`RUN_REPORT_FILE`
`RUN_LOGS_DIR`
`NOTES_FILE`
`NOTE_MARK`
`LOCAL_HOST_NAME`

Exceptions

ExperimentError

Raised when an experiment cannot be read, started or driven.

Functions

`read_manifest`(...)	Ask the module what it takes, and return what it said.
`read_devices`(→ list[str] \| None)	Ask the module which devices of its `setup.yaml` a run of config uses.
`_ask_module`(→ subprocess.CompletedProcess[str])	Run a question the module answers without connecting, and capture the answer.
`run_environment`(→ dict[str, str])	Build the environment a run gets.
`runs_of`(→ list[pathlib.Path])	Return this experiment's run directories, newest first.
`check_run_id`(→ str)	Return text if it is usable as a run directory name, else raise.
`created_of`(→ dict[str, Any])	Return what the console recorded when the run was created, if anything.
`add_note`(→ None)	Append a note to the run's `NOTES_FILE`, while it runs or after it ended.
`notes_of`(→ list[dict[str, str]])	The run's notes, oldest first: each `{"at", "by", "text"}`; none without a file.
`create_run`(→ dict[str, Any])	Name a run before it is started, recording what it will be run with.
`has_started`(→ bool)	Whether the run has started: its configuration is frozen into its plan then.
`rename_run`(→ pathlib.Path)	Give a run that has not started another name, which is its directory's.
`reconfigure_run`(→ pathlib.Path)	Replace the configuration of a run that has not started.
`report_of`(→ dict[str, Any] \| None)	The run's own report, or `None` when it has none.
`keep_log`(→ pathlib.Path)	Copy a job's log into the run it ran, under `RUN_LOGS_DIR`.
`kept_log`(→ pathlib.Path \| None)	The newest job log run_dir keeps, or `None` when it keeps none.
`latest`(→ pathlib.Path \| None)	Return the newest run of run_root, or `None` when there is none.
`find_run`(→ pathlib.Path \| None)	Return the run directory run_id names, or the newest one.
`resumable`(→ str \| None)	Return the id of the newest run that could be continued, if any.
`status_of`(→ dict[str, Any])	Return what a run's journal says about it, by replaying the journal here.
`_replay_status`(→ dict[str, Any])	Derive a run's status from its journal.
`progress_of`(→ dict[str, Any] \| None)	Return what the step in flight last said about itself.
`steps_of`(→ list[dict[str, Any]])	Return the plan's steps, so a status can name the one in flight.
`records_of`(→ list[dict[str, Any]])	Return the newest committed step records.
`ask`(→ dict[str, Any])	Amend a run's control file and return what it now asks for.
`asked`(→ dict[str, Any])	Return what a run is currently being asked for.
`clear`(→ None)	Remove a run's control file, so a resumed run is asked for nothing.
`reports_in`(→ list[str])	Collect the operator's reports out of run statuses, in order.

Module Contents

plesty.server.services.experiments.MANIFEST_TIMEOUT = 60.0

plesty.server.services.experiments.JOB_KIND = 'experiment'

plesty.server.services.experiments.VIEWER_KIND = 'viewer'

plesty.server.services.experiments.CONTROL_FILE = 'control.json'

plesty.server.services.experiments.CANCEL = 'cancel'

plesty.server.services.experiments.ABORT = 'abort'

plesty.server.services.experiments.CREATED_FILE = 'created.json'

plesty.server.services.experiments.RUN_CONFIG_FILE = 'config.yaml'

plesty.server.services.experiments.RUN_REPORT_FILE = 'report.html'

plesty.server.services.experiments.RUN_LOGS_DIR = 'job-logs'

plesty.server.services.experiments.NOTES_FILE = 'notes.txt'

plesty.server.services.experiments.NOTE_MARK = '--- '

exception plesty.server.services.experiments.ExperimentError

Bases: RuntimeError

Raised when an experiment cannot be read, started or driven.

Initialize self. See help(type(self)) for accurate signature.

plesty.server.services.experiments.read_manifest(spec: plesty.server.model.experiment.ExperimentSpec, python: str | pathlib.Path, cwd: str | pathlib.Path, env: Mapping[str, str] | None = None, timeout: float = MANIFEST_TIMEOUT) → plesty.server.model.experiment.ExperimentManifest

Ask the module what it takes, and return what it said.

Parameters:

spec (plesty.server.model.experiment.ExperimentSpec) – The declared experiment.
python (str | pathlib.Path) – Interpreter of its installed environment.
cwd (str | pathlib.Path) – Where to run it — the checkout, so a relative configuration path in the answer means what it says.
env (Mapping[str, str] | None) – Environment for the call; the manifest needs none, but a module that reads its own package data may.
timeout (float) – Seconds to wait.

Returns:

The parsed manifest.

Raises:

ExperimentError – If the command fails, times out, or prints something that is not a manifest.

Return type:

plesty.server.model.experiment.ExperimentManifest

plesty.server.services.experiments.read_devices(spec: plesty.server.model.experiment.ExperimentSpec, python: str | pathlib.Path, cwd: str | pathlib.Path, config: str, env: Mapping[str, str] | None = None, timeout: float = MANIFEST_TIMEOUT) → list[str] | None

Ask the module which devices of its setup.yaml a run of config uses.

Only for a module whose manifest says chooses_devices (plesty-lib#62).

Parameters:

spec (plesty.server.model.experiment.ExperimentSpec) – The declared experiment.
python (str | pathlib.Path) – Interpreter of its installed environment.
cwd (str | pathlib.Path) – The checkout.
config (str) – The configuration file the run will read.
env (Mapping[str, str] | None) – Environment for the call.
timeout (float) – Seconds to wait.

Returns:

The device names, or None for all of them.

Raises:

ExperimentError – If the command fails, times out, or prints no list.

Return type:

list[str] | None

plesty.server.services.experiments._ask_module(spec: plesty.server.model.experiment.ExperimentSpec, command: list[str], cwd: str | pathlib.Path, env: Mapping[str, str] | None, timeout: float) → subprocess.CompletedProcess[str]

Run a question the module answers without connecting, and capture the answer.

Raises:

ExperimentError – If it could not be run, or did not answer in time.

Parameters:

spec (plesty.server.model.experiment.ExperimentSpec)
command (list[str])
cwd (str | pathlib.Path)
env (Mapping[str, str] | None)
timeout (float)

Return type:

subprocess.CompletedProcess[str]

plesty.server.services.experiments.LOCAL_HOST_NAME = 'local'

plesty.server.services.experiments.run_environment(spec: plesty.server.model.experiment.ExperimentSpec, base: Mapping[str, str], dotenv: Mapping[str, str]) → dict[str, str]

Build the environment a run gets.

The bench’s own environment first, then the module’s .env, then what the fleet declares — so the declaration is what an operator edits to point one experiment at a different stage, and the .env stays the module’s own deployment file.

Parameters:

spec (plesty.server.model.experiment.ExperimentSpec) – The declared experiment.
base (Mapping[str, str]) – The bench’s environment.
dotenv (Mapping[str, str]) – The module’s .env, already parsed.

Returns:

The environment for the run.

Return type:

dict[str, str]

plesty.server.services.experiments.runs_of(run_root: str | pathlib.Path) → list[pathlib.Path]

Return this experiment’s run directories, newest first.

A run directory is one holding a journal.jsonl — or a created.json, for a run named and configured but not yet started. The run root may also hold whatever the module wrote beside its runs, and a directory that is neither is not a run anything can report on.

Parameters:: run_root (str | pathlib.Path) – The experiment’s run root.
Returns:: The run directories, newest first by name — run ids carry a date, so name order is close enough to time order to read as it.
Return type:: list[pathlib.Path]

plesty.server.services.experiments.check_run_id(text: str) → str

Return text if it is usable as a run directory name, else raise.

Deliberately the console’s own check rather than plesty-lib’s, for the same reason the journal is replayed here: the rule is about what a directory name may be, and the console must apply it on any plesty-lib — including one too old to carry the check at all. It is the same rule, and a short one.

Parameters:: text (str) – The proposed run id.
Returns:: The same string, trimmed.
Raises:: ExperimentError – If it is empty, a path, or holds a character that does not belong in a directory name.
Return type:: str

plesty.server.services.experiments.created_of(run_dir: str | pathlib.Path) → dict[str, Any]

Return what the console recorded when the run was created, if anything.

Parameters:: run_dir (str | pathlib.Path) – The run directory.
Returns:: The document, or {} for a run that was started directly.
Return type:: dict[str, Any]

plesty.server.services.experiments.add_note(run_dir: str | pathlib.Path, text: str, by: str, when: Any = None) → None

Append a note to the run’s NOTES_FILE, while it runs or after it ended.

Parameters:

run_dir (str | pathlib.Path) – The run directory.
text (str) – The note; blank lines inside it are kept.
by (str) – Who wrote it.
when (Any) – A datetime; now by default (tests pass one).

Raises:

ExperimentError – When the note is empty or cannot be written.

Return type:

None

plesty.server.services.experiments.notes_of(run_dir: str | pathlib.Path) → list[dict[str, str]]

The run’s notes, oldest first: each {"at", "by", "text"}; none without a file.

Parameters:: run_dir (str | pathlib.Path)
Return type:: list[dict[str, str]]

plesty.server.services.experiments.create_run(run_dir: str | pathlib.Path, config_text: str | None = None, **fields: Any) → dict[str, Any]

Name a run before it is started, recording what it will be run with.

Parameters:

run_dir (str | pathlib.Path) – The directory the run will use — its name is the run id.
config_text (str | None) – The run’s own configuration, written as its RUN_CONFIG_FILE; none is written when None.
**fields (Any) – What to record: the experiment, the configuration file the run’s copy came from, and whatever else the run needs to be started later.

Returns:

The document written.

Raises:

ExperimentError – If the directory already holds a run, or cannot be written. Naming a second run over a started one would interleave two journals into one directory.

Return type:

dict[str, Any]

plesty.server.services.experiments.has_started(run_dir: str | pathlib.Path) → bool

Whether the run has started: its configuration is frozen into its plan then.

Parameters:: run_dir (str | pathlib.Path)
Return type:: bool

plesty.server.services.experiments.rename_run(run_dir: str | pathlib.Path, new_name: str) → pathlib.Path

Give a run that has not started another name, which is its directory’s.

The directory moves with everything in it — the run’s configuration and its CREATED_FILE — and a duplicate recorded as a copy of it follows the new name, so it stays beside it in the table (#27).

Parameters:

run_dir (str | pathlib.Path) – The run directory.
new_name (str) – What to call it; checked as check_run_id() checks.

Returns:

The run’s directory under its new name.

Raises:

ExperimentError – If the name is unusable or taken, the run has started — its journal and data are written under the name it has — or the directory cannot be moved.

Return type:

pathlib.Path

plesty.server.services.experiments.reconfigure_run(run_dir: str | pathlib.Path, config_text: str, config: str) → pathlib.Path

Replace the configuration of a run that has not started.

Parameters:

run_dir (str | pathlib.Path) – The run directory.
config_text (str) – The run’s configuration, as it is to be run.
config (str) – The module’s file it was taken from (config/demo.yaml), recorded in CREATED_FILE so the run table names it.

Returns:

The run’s configuration file.

Raises:

ExperimentError – If the run has started — its plan was built from the configuration it had, and resuming checks it against that — or the files cannot be written.

Return type:

pathlib.Path

plesty.server.services.experiments.report_of(run_dir: str | pathlib.Path) → dict[str, Any] | None

The run’s own report, or None when it has none.

Parameters:: run_dir (str | pathlib.Path) – The run directory.
Returns:: Its name, path, bytes and written (epoch seconds), or None when no report has been written — a run that has not ended, or one that ended before this bench wrote them.
Return type:: dict[str, Any] | None

plesty.server.services.experiments.keep_log(run_dir: str | pathlib.Path, log: str | pathlib.Path) → pathlib.Path

Copy a job’s log into the run it ran, under RUN_LOGS_DIR.

Copied when the job ends, not written there from the start: a run the module names itself has no directory until it has begun, and the job’s log is already being written by then.

Parameters:

run_dir (str | pathlib.Path) – The run directory.
log (str | pathlib.Path) – The job’s log, named after the job.

Returns:

The copy.

Raises:

OSError – When the log cannot be read or the copy written.

Return type:

pathlib.Path

plesty.server.services.experiments.kept_log(run_dir: str | pathlib.Path) → pathlib.Path | None

The newest job log run_dir keeps, or None when it keeps none.

Job ids end in their start time, so the newest is the last by name — the one a resumed run was finished by.

Parameters:: run_dir (str | pathlib.Path)
Return type:: pathlib.Path | None

plesty.server.services.experiments.latest(run_root: str | pathlib.Path) → pathlib.Path | None

Return the newest run of run_root, or None when there is none.

Parameters:: run_root (str | pathlib.Path)
Return type:: pathlib.Path | None

plesty.server.services.experiments.find_run(run_root: str | pathlib.Path, run_id: str | None) → pathlib.Path | None

Return the run directory run_id names, or the newest one.

Parameters:

run_root (str | pathlib.Path) – The experiment’s run root.
run_id (str | None) – The run to find, or None for the newest.

Returns:

The directory, or None when it does not exist. A run created but not yet started counts: it is the one an operator is about to run.

Return type:

pathlib.Path | None

plesty.server.services.experiments.resumable(run_root: str | pathlib.Path) → str | None

Return the id of the newest run that could be continued, if any.

A run is resumable when it stopped without finishing: canceled or aborted, or left running by a process that is no longer there. A completed run is not, and neither is one that never wrote a journal.

Parameters:: run_root (str | pathlib.Path) – The experiment’s run root.
Returns:: The run id, or None.
Return type:: str | None

plesty.server.services.experiments.status_of(run_dir: str | pathlib.Path) → dict[str, Any]

Return what a run’s journal says about it, by replaying the journal here.

Deliberately not plesty.lib.experiment.Journal.status(), though it computes the same thing. The run was written by an experiment module, whose plesty-lib has nothing to do with the one this console happens to have installed: delegating would make the same run read differently on two benches, and a run written by a newer module would lose its paused state on an older console. The journal is an append-only document with five event names, and reading it is cheaper than the coupling.

Parameters:: run_dir (str | pathlib.Path) – The run directory.
Returns:: state, completed, failed, reports, started, ended, provenance and the last event. A run that was named but not yet started is created, which is distinct from not_started: one is a run waiting to go, the other an experiment that has never had one.
Return type:: The status mapping

plesty.server.services.experiments._replay_status(journal: pathlib.Path) → dict[str, Any]

Derive a run’s status from its journal.

Parameters:: journal (pathlib.Path) – The journal file.
Returns:: The status mapping.
Return type:: dict[str, Any]

plesty.server.services.experiments.progress_of(run_dir: str | pathlib.Path) → dict[str, Any] | None

Return what the step in flight last said about itself.

Parameters:: run_dir (str | pathlib.Path) – The run directory.
Returns:: The reported progress, or None when the step reports none. A step that says nothing is not an error — most are short enough not to.
Return type:: dict[str, Any] | None

plesty.server.services.experiments.steps_of(run_dir: str | pathlib.Path) → list[dict[str, Any]]

Return the plan’s steps, so a status can name the one in flight.

Parameters:: run_dir (str | pathlib.Path) – The run directory.
Returns:: One {id, op, params} per step; empty when the plan is unreadable.
Return type:: list[dict[str, Any]]

plesty.server.services.experiments.records_of(run_dir: str | pathlib.Path, limit: int = 200) → list[dict[str, Any]]

Return the newest committed step records.

Parameters:

run_dir (str | pathlib.Path) – The run directory.
limit (int) – How many to return, newest last.

Returns:

The records in write order. A record is the durable result of its step, so this is the raw data a run has produced so far.

Return type:

list[dict[str, Any]]

plesty.server.services.experiments.ask(run_dir: str | pathlib.Path, **changes: Any) → dict[str, Any]

Amend a run’s control file and return what it now asks for.

The file is replaced rather than rewritten, so a run reading it sees either the old document or the new one and never half of each.

Parameters:

run_dir (str | pathlib.Path) – The run directory.
**changes (Any) – Fields of the control document to set.

Returns:

The control document as it now stands.

Raises:

ExperimentError – If the file cannot be written.

Return type:

dict[str, Any]

plesty.server.services.experiments.asked(run_dir: str | pathlib.Path) → dict[str, Any]

Return what a run is currently being asked for.

Parameters:: run_dir (str | pathlib.Path) – The run directory.
Returns:: The control document; empty when nothing is being asked.
Return type:: dict[str, Any]

plesty.server.services.experiments.clear(run_dir: str | pathlib.Path) → None

Remove a run’s control file, so a resumed run is asked for nothing.

A run resumed while its control file still says cancel would stop at its first step, which reads as a resume that did nothing.

Parameters:: run_dir (str | pathlib.Path) – The run directory.
Return type:: None

plesty.server.services.experiments.reports_in(*documents: Mapping[str, Any]) → list[str]

Collect the operator’s reports out of run statuses, in order.

Parameters:: *documents (Mapping[str, Any]) – Status mappings.
Returns:: Every report line.
Return type:: list[str]