Record and select a training run

Problem

When you need this. Two jobs can use the same code and data yet represent different executions. One completes; the other times out after writing half a report. If a later step simply opens latest/evaluation.json, it can select unfinished work or a file replaced after review.

The setup-capture lesson established what the trainer would use. Now you will attach an execution record to that setup and select its output explicitly. The lab reuses the same small CPU trainer and bundled Rice dataset. It adds lifecycle and artifact handling, without changing the learning algorithm.

Concepts

A setup identity describes what a job is intended to use. A run identity names one execution attempt, including a failed or timed-out attempt. An artifact identity identifies particular output bytes produced by an eligible attempt.

Process exit, report validity and task quality are different checks. The wrapper accepts a report only when the process and schema/setup checks pass. That report contains evaluation results, not the complete weights and optimizer state needed to resume training. Release-quality policy and resumable checkpoints are separate work.

Architecture

Execution and selection are separate operations. A failed or timed-out run keeps its diagnostics, while selection accepts only a succeeded run whose archived bytes match the retained expectation.

The wrapper verifies the chosen setup, copies and checks the retained inputs, then launches the existing trainer in a new run directory. It captures exit status and logs, validates the required report and binds eligible output to the run. Selection verifies the expected record and artifact identities without following a mutable latest alias.

Working code

Inspect one completed run before starting another

Download the complete lab and enter mlops-run-artifacts/. Use Python 3.12 or later:

Prepare the lab and inspect the saved runTerminal · bash

python3.12 -m venv .venv
source .venv/bin/activate
python inspect_runs.py
python -m unittest -v test_runs.py

There are no third-party packages to install. These commands inspect the bundled run and exercise offline failure tests; they do not start training.

Recorded inspection of the saved local CPU runRecorded output · textDownload

Saved run: run-a6dc7a75af3349d7ad000fef92961c88
State transitions: planned -> running -> succeeded
Selected artifact: training_evaluation_report
Report bytes: 2143; SHA-256 verified
Saved validation: 590 rows; accuracy 0.9271
Edited report: refused; mutable latest alias: refused
Inspection complete; no training started

The saved job actually executed the baseline for 200 epochs. Its report contains validation metrics, including accuracy of 0.9271 across 590 validation rows. The original baseline’s records remain unchanged. This invocation did not use --final or score the held-out test partition; the split membership counts in stdout are separate from test metrics.

The artifact type is training_evaluation_report. The unchanged trainer rounds its model and scaler values to four decimals when writing that report. Treat it as a record of the experiment, not a full-precision checkpoint to serve or resume.

Give setup, execution and output different identities

The lab keeps three identities because they answer different questions:

Identity	Question it answers	Where it is used
Setup ID	Which code, inputs, configuration and runtime were captured?	Checked before launch and after copying inputs.
Run ID	Which execution attempt is this?	A fresh UUID names an exclusively created directory.
Report SHA-256	Which exact output bytes were registered?	Stored inside the completed run record.

The final run record also has a SHA-256 digest. The selection record printed by run gives you that digest and the run ID. Retain both separately: they are the selection expectation, rather than values to recalculate from files that may have changed.

Here, immutable selection means that the wrapper does not overwrite an allocated run ID and the selector refuses changed bytes against that expectation. Ordinary files remain editable. Digests do not authenticate an author or protect a reference that an attacker can also replace.

Verify the copied setup before launching the process

execute() first verifies the captured setup. It allocates a new run directory, records planned, and copies the four fixed inputs into workspace/. It verifies that copy before moving to running and launching the trainer.

This second check matters: copying an input is another opportunity for the execution inputs to diverge from the captured setup. The trainer runs from its copied workspace, so a later edit to the lab’s source directory does not silently change this job’s input files. The exercise assumes a trusted local workspace during execution; it does not provide process isolation against hostile concurrent edits.

Inspect job execution and artifact selection

run_jobs.pySource · pythonDownload

"""Run one supported local validation job; select only digest-bound completed reports."""
import argparse
from datetime import datetime, timezone
import hashlib
import json
import math
from pathlib import Path
import re
import shutil
import subprocess
import sys
import time
import uuid
from vendor import capture_job

HERE = Path(__file__).resolve().parent
RUN_ID = re.compile(r'run-[0-9a-f]{32}\Z')
SHA = re.compile(r'[0-9a-f]{64}\Z')
ARTIFACT_KIND = 'training_evaluation_report'


def digest(raw):
    return hashlib.sha256(raw).hexdigest()


def now():
    return datetime.now(timezone.utc).isoformat(timespec='microseconds')


def write_state(folder, record):
    # Atomic replacement of this run's state file, not a crash-recovery service.
    path = folder / 'run.json.next'
    with path.open('x') as file:
        file.write(json.dumps(record, indent=2, allow_nan=False) + '\n')
    path.replace(folder / 'run.json')


def launch(argv, folder, timeout, stdout, stderr):
    return subprocess.run(argv, cwd=folder, timeout=timeout,
                          stdout=stdout, stderr=stderr, check=False)


def validate_report(raw, snapshot):
    report = capture_job.parse_config(raw)
    if not isinstance(report, dict) or not {'data', 'settings', 'model', 'validation'} <= set(report):
        raise ValueError('missing training-report fields')
    expected = snapshot['effective_config']
    try:
        if not all(isinstance(report[name], dict) for name in ('data', 'settings', 'model', 'validation')):
            raise ValueError('report sections must be objects')
        settings = report['settings']
        if ('test' in report or report['data'].get('sha256') != snapshot['files']['trainer/rice.csv']
                or type(settings.get('epochs')) is not int
                or settings['epochs'] != expected['epochs']
                or type(settings.get('learning_rate')) not in (int, float)
                or settings['learning_rate'] != expected['learning_rate']
                or settings.get('positive_class') != expected['positive_class']
                or settings.get('scaler_fitted_on') != 'training'
                or report['split']['salt'] != expected['split_salt']):
            raise ValueError('report does not match validation setup')
        metric = report['validation']['logistic']
        if (not isinstance(metric, dict) or type(metric['rows']) is not int or metric['rows'] < 1
                or type(metric['accuracy']) not in (int, float)
                or not math.isfinite(metric['accuracy']) or not 0 <= metric['accuracy'] <= 1):
            raise ValueError('invalid validation metric fields')
    except (KeyError, TypeError) as error:
        raise ValueError('malformed training report') from error
    return report


def execute(runs, snapshot, expected_recipe, root=HERE, timeout=30):
    if type(timeout) is not int or not 1 <= timeout <= 30:
        raise ValueError('timeout must be an integer from 1 to 30 seconds')
    capture_job.verify(snapshot, expected_recipe, root)
    if snapshot['effective_config']['mode'] != 'validation':
        raise ValueError('this job wrapper runs validation only, never --final')
    runs = Path(runs)
    runs.mkdir(parents=True, exist_ok=True)
    run_id = 'run-' + uuid.uuid4().hex
    folder = runs / run_id
    folder.mkdir()  # Exclusive allocation: there is no reuse/overwrite path.
    workspace = folder / 'workspace'
    workspace.mkdir()
    setup_raw = (json.dumps(snapshot, indent=2) + '\n').encode()
    (folder / 'setup.json').write_bytes(setup_raw)
    record = {'format': 1, 'run_id': run_id, 'state': 'planned',
              'setup_id': expected_recipe, 'setup_sha256': digest(setup_raw),
              'transitions': ['planned'],
              'created_at': now(), 'started_at': None, 'finished_at': None,
              'returncode': None, 'timeout_seconds': timeout, 'reason': '', 'artifact': None,
              'launcher_sha256': digest((HERE / 'run_jobs.py').read_bytes()),
              'argv': ['<sys.executable>', 'workspace/trainer/baseline.py', '--out', 'evaluation.partial.json']}
    write_state(folder, record)
    start = time.monotonic()
    try:
        for name in capture_job.FILES:
            target = workspace / name
            target.parent.mkdir(parents=True, exist_ok=True)
            shutil.copyfile(Path(root) / name, target)
        capture_job.verify(snapshot, expected_recipe, workspace)
        record.update(state='running', started_at=now())
        record['transitions'].append('running')
        write_state(folder, record)
        argv = [sys.executable, *record['argv'][1:]]
        with (folder/'stdout.txt').open('xb') as stdout, (folder/'stderr.txt').open('xb') as stderr:
            result = launch(argv, folder, timeout, stdout, stderr)
        record['returncode'] = result.returncode
        if result.returncode != 0:
            raise ValueError('trainer exited with a nonzero status')
        raw = (folder/'evaluation.partial.json').read_bytes()
        validate_report(raw, snapshot)
        final = folder/'evaluation.json'
        with final.open('xb') as file:
            file.write(raw)
        (folder/'evaluation.partial.json').unlink()
        record.update(state='succeeded', artifact={'kind': ARTIFACT_KIND,
                      'file': 'evaluation.json', 'sha256': digest(raw), 'bytes': len(raw)})
    except subprocess.TimeoutExpired:
        record.update(state='timed_out', reason='trainer exceeded the process timeout')
    except (OSError, ValueError, KeyError, TypeError) as error:
        record.update(state='failed', reason=type(error).__name__ + ': ' + str(error))
    record['transitions'].append(record['state'])
    record.update(finished_at=now(), elapsed_seconds=round(time.monotonic()-start, 6))
    record['logs'] = {name: digest((folder/name).read_bytes())
                      for name in ('stdout.txt','stderr.txt') if (folder/name).is_file()}
    write_state(folder, record)
    return {'run_id': run_id, 'record_sha256': digest((folder/'run.json').read_bytes()),
            'state': record['state'], 'artifact': record['artifact']}


def regular_file(path):
    if path.is_symlink() or not path.is_file():
        raise ValueError('selection needs a regular file inside the run directory')
    return path.read_bytes()


def select(runs, run_id, expected_record_sha256):
    if not isinstance(run_id, str) or not RUN_ID.fullmatch(run_id):
        raise ValueError('select an explicit run ID; aliases and paths are not accepted')
    if not isinstance(expected_record_sha256, str) or not SHA.fullmatch(expected_record_sha256):
        raise ValueError('selection requires the expected run-record SHA-256')
    folder = Path(runs) / run_id
    if folder.is_symlink() or not folder.is_dir():
        raise ValueError('run directory is missing or substituted')
    raw = regular_file(folder/'run.json')
    if digest(raw) != expected_record_sha256:
        raise ValueError('run record differs from the selected identity')
    record = capture_job.parse_config(raw)
    if not isinstance(record, dict) or record.get('run_id') != run_id or record.get('state') != 'succeeded':
        raise ValueError('only the selected succeeded run has an eligible artifact')
    setup_raw = regular_file(folder/'setup.json')
    snapshot = capture_job.parse_config(setup_raw)
    if digest(setup_raw) != record['setup_sha256'] or capture_job.identity(snapshot) != record['setup_id']:
        raise ValueError('setup snapshot binding differs')
    for name, expected in snapshot['files'].items():
        if name not in capture_job.FILES or digest(regular_file(folder/'workspace'/name)) != expected:
            raise ValueError('archived input bytes differ from setup')
    if set(snapshot['files']) != set(capture_job.FILES):
        raise ValueError('archived input inventory differs')
    for name, expected in record['logs'].items():
        if name not in ('stdout.txt', 'stderr.txt') or digest(regular_file(folder/name)) != expected:
            raise ValueError('archived log bytes differ from run record')
    artifact = record['artifact']
    if (not isinstance(artifact, dict) or artifact.get('kind') != ARTIFACT_KIND
            or artifact.get('file') != 'evaluation.json'):
        raise ValueError('unexpected artifact type or filename')
    report_raw = regular_file(folder/'evaluation.json')
    if digest(report_raw) != artifact['sha256'] or len(report_raw) != artifact['bytes']:
        raise ValueError('artifact bytes differ from the selected run record')
    report = validate_report(report_raw, snapshot)
    return {'run_id': run_id, 'record_sha256': expected_record_sha256,
            'artifact': artifact, 'setup_id': record['setup_id'],
            'validation': report['validation']['logistic']}


def main():
    parser = argparse.ArgumentParser()
    commands = parser.add_subparsers(dest='command', required=True)
    run = commands.add_parser('run')
    run.add_argument('--runs', type=Path, required=True)
    run.add_argument('--timeout', type=int, default=30)
    choose = commands.add_parser('select')
    choose.add_argument('--runs', type=Path, required=True)
    choose.add_argument('--run-id', required=True)
    choose.add_argument('--record-sha256', required=True)
    args = parser.parse_args()
    if args.command == 'run':
        setup = capture_job.capture(HERE)
        result = execute(args.runs, setup, capture_job.identity(setup), timeout=args.timeout)
    else:
        result = select(args.runs, args.run_id, args.record_sha256)
    print(json.dumps(result, indent=2))
    if args.command == 'run' and result['state'] != 'succeeded':
        raise SystemExit(1)


if __name__ == '__main__':
    main()

To execute an optional new validation job:

Start a new local CPU validation runTerminal · bash

python run_jobs.py run --runs my-runs --timeout 30

This command performs real local training. It needs no network, model API or GPU. The wrapper uses the invoking Python interpreter and an argument list, with no shell. The archived argument list represents that interpreter as <sys.executable> instead of publishing an installation path.

The process timeout is limited to 1–30 seconds. Python’s subprocess documentation explains that run() kills and waits for a child when its timeout expires; process creation can delay when the timeout is reported. This is not a deadline covering every operation in the wrapper.

Promote a report only after checking the result

A zero exit is necessary, but a missing or truncated JSON report is still a failed job. The wrapper checks the report’s dataset identity, effective settings, split salt and validation metric fields. Only then does it register evaluation.json, its byte count and its digest, and finish with succeeded.

A nonzero exit or invalid report finishes as failed. A process timeout finishes as timed_out. Both retain stdout, stderr and any partial report for diagnosis, but neither has an eligible artifact. The output checks establish the wrapper’s report contract; they do not prove that the labels or learning algorithm are scientifically correct.

State updates replace run.json through a temporary file. This is a local state transition, not a crash-recovery service. If the wrapper itself crashes, an old planned or running record can remain. Selection refuses that state rather than inferring success from a report file’s presence.

Select the recorded bytes, rather than a moving alias

Use the ID and record digest printed by your new run command. Replace both uppercase placeholders below with those values:

Select one explicit completed runTerminal · bash

python run_jobs.py select --runs my-runs \
  --run-id RUN_ID_FROM_THE_RUN_COMMAND \
  --record-sha256 RECORD_SHA256_FROM_THE_RUN_COMMAND

The selector first checks the expected run-record digest and succeeded state. It then checks the archived setup, fixed input copy, logs and evaluation report against their recorded identities. Inspecting an old artifact does not require today’s interpreter to match the training interpreter; those runtime fields describe the archived job.

latest is refused because it is not a run ID. Adding even a newline to the report is also refused: it preserves the parsed JSON value but changes the selected bytes. The inspection driver demonstrates both boundaries using a temporary copy:

Inspect the saved-run selection experiment

inspect_runs.pySource · pythonDownload

"""Inspect the saved local run and exercise selection refusals without retraining."""
import argparse
import json
from pathlib import Path
import shutil
import tempfile
import run_jobs as job

HERE = Path(__file__).resolve().parent


def inspect():
    receipt = json.loads((HERE/'recorded-selection.json').read_text())
    selected = job.select(HERE/'recorded-runs', receipt['run_id'], receipt['record_sha256'])
    folder = HERE/'recorded-runs'/receipt['run_id']
    record = json.loads((folder/'run.json').read_text())
    if record['launcher_sha256'] != job.digest((HERE/'run_jobs.py').read_bytes()):
        raise ValueError('current launcher differs from the recorded execution source')
    refusals = {}
    with tempfile.TemporaryDirectory() as tmp:
        runs = Path(tmp)/'runs'
        shutil.copytree(HERE/'recorded-runs', runs)
        artifact = runs/receipt['run_id']/'evaluation.json'
        artifact.write_bytes(artifact.read_bytes()+b'\n')
        for name, run_id in (('edited_report', receipt['run_id']), ('mutable_alias', 'latest')):
            try:
                job.select(runs, run_id, receipt['record_sha256'])
            except ValueError as error:
                refusals[name] = str(error)
            else:
                raise AssertionError(name+' was accepted')
    return {'execution_mode':'inspection of a saved local CPU validation run; no new training',
            'selected':selected, 'transitions':record['transitions'],
            'returncode':record['returncode'], 'refusals':refusals}


def main():
    parser=argparse.ArgumentParser()
    parser.add_argument('--save',type=Path)
    parser.add_argument('--check',action='store_true')
    args=parser.parse_args()
    result=inspect()
    if args.check:
        if result != json.loads((HERE/'result.json').read_text()):
            raise SystemExit('saved inspection differs')
    if args.save:
        with args.save.open('x') as file:
            file.write(json.dumps(result,indent=2)+'\n')
    selected=result['selected']
    print('Saved run: '+selected['run_id'])
    print('State transitions: '+' -> '.join(result['transitions']))
    print('Selected artifact: '+selected['artifact']['kind'])
    print('Report bytes: '+str(selected['artifact']['bytes'])+'; SHA-256 verified')
    metric=selected['validation']
    print(f"Saved validation: {metric['rows']} rows; accuracy {metric['accuracy']:.4f}")
    print('Edited report: refused; mutable latest alias: refused')
    print('Inspection complete; no training started')


if __name__=='__main__': main()

python inspect_runs.py --check compares this inspection with the bundled record. It performs no new training. The complete run, setup and report are included in the lab ZIP, alongside the separately retained recorded-selection.json selection record.

The 20 offline tests use explicit process doubles for exit, timeout and output failures. They also cover setup drift before launch, changes during copying, duplicate run allocation, strict JSON parsing and substituted archive bytes. Synthetic metric fields in those unit tests are test inputs, not additional measured training runs.

Design exercise: a job exits successfully but its report is missing. Which state and artifact should the wrapper record? Then consider a reviewer selecting an older completed run after the local Python version changes. Explain why execution verification enforces runtime identity, while archived artifact selection can inspect the old run without executing it.

Compare your reasoning

A zero exit with no required report finishes as failed and has no eligible artifact; retain the attempt and its logs. Selecting an older completed run verifies archived bytes against the retained expectation without running the trainer, so it can inspect a different recorded Python version. Launching a new run instead checks the current runtime against its selected setup because that environment will execute the code.

Use the selected evaluation report as input to a later model-quality decision. A new attempt gets a new run ID. This wrapper does not record a parent-run link or restore a training checkpoint; neither can be inferred from a directory name.

Failure cases

A process can exit zero without creating its required report. A timed-out process can leave partial output. Both attempts remain inspectable but cannot supply an eligible artifact. After a successful run, changing a report, log or archived input must invalidate selection rather than silently redefine the reviewed attempt.

Check the input and failure boundaries

Trigger	Required behavior
Nonzero process exit	Retain the failed attempt and do not promote its report.
Timeout	Retain timeout logs without an eligible output artifact.
No output despite zero exit	Mark the attempt failed when the required report is absent.
Malformed report	Reject an output that does not satisfy the report schema.
Setup and report differ	Refuse output whose setup fields or mode differ from the job contract.
Setup drift before launch	Stop before launching when selected setup verification fails.
Copied input drift	Verify the copied inputs before invoking the child process.
Final test metrics requested	Refuse the final-test mode for this development workflow.
Mutable alias or path	Require a content-selected artifact instead of a mutable alias or path.
Run record changed	Reject changed run-record bytes against the expected identity.
Artifact changed	Reject changed artifact bytes even when JSON meaning appears unchanged.
Input archive changed	Keep the selected setup and archived input bindings intact.
Logs changed	Reject substituted log bytes.
Duplicate run ID	Refuse an existing run ID instead of overwriting its attempt.
Symlink substitution	Reject a symlink substituted for an expected retained file.

Deployment

Run each job attempt in a new directory with an explicit setup identity and execution mode. Retain failed logs for diagnosis and publish an artifact reference only after the wrapper’s checks pass. Downstream release code must select the expected immutable reference and apply its own quality policy; this local wrapper does not deploy a model.

Scaling

Parallel jobs need collision-resistant run identities, isolated working directories and enough storage for retained inputs and logs. A later scheduler or object store must preserve the same selection contract. Timeout and process supervision also need to cover the actual job process tree before transferring this local wrapper to another execution environment.

Key takeaways

A setup can have several execution attempts with different outcomes.
Successful process exit is necessary but insufficient for an eligible report.
An immutable evaluation report is not a resumable model checkpoint.

Read the explanation

Zero exit is necessary but insufficient. The required report must exist and pass validation before registration.

Best practices

Keep failed attempts instead of replacing them with the next run.
Select exact output identity rather than latest aliases.
Retain logs and setup bindings with the report they explain.

References

Python, subprocess.run: process invocation, completion and timeout behavior.
Trainer setup capture: effective defaults, file identity and runtime descriptors used by this wrapper.

Sources checked October 24, 2026. The saved result comes from one local CPU validation run; the inspection commands reuse that record.