Storage#

On-disk job folder protocol: identity-addressed outputs, with provenance.

Layout under the orchestrator’s root:

<root>/<job_key>/
    manifest.json   parameters, callable location, provenance, timings
    deps.json       the environment fingerprint
    metrics/*.csv   append-only (step, value), one file per metric
    DONE | FAILED   completion marker; FAILED carries the traceback

A job’s state is read from its marker rather than from a mutable field, and a job is built in a staging directory that is renamed into place only once its marker is written. A job killed mid-flight therefore leaves no marker at the final path and is simply “not done” on the next pass: there is no folder that is incomplete yet marked complete, and no lock to reason about.

See notes/adr-2026-10-02-orchestrator-v2-a-design.md.

class krum.orchestration.storage.JobFolder(path: str | PathLike[str])[source]#

Bases: object

One job’s directory on disk, and the questions one can ask of it.

called() → dict[str, str] | None[source]#

The functions this job entered, or None if it was not traced.

An untraced job is not the same as a job that called nothing, so the two are told apart: there is nothing to verify in the first case.

deps() → dict[str, Any][source]#: Read the job’s fingerprint.

property done: bool#: Whether this job completed successfully.

property key: str#: The job key naming this directory.

manifest() → dict[str, Any][source]#: Read the job’s manifest.

metric_names() → list[str][source]#: The metrics this job recorded, sorted.

metric_path(name: str) → Path[source]#: The file holding one metric’s rows, whether or not it exists.

metric_rows(name: str) → int | None[source]#

How many rows this job recorded for a metric, if it says.

Read from the manifest rather than from the file, so asking costs nothing. None for a job written before the count was recorded.

property path: Path#: The job directory.

property status: str#: One of done, failed or absent, read from the markers.

traceback() → str | None[source]#: The recorded traceback of a failed job, if any.

witnesses() → dict[str, Any][source]#: The witnesses recorded with this job, empty if it has none.

class krum.orchestration.storage.JobStore(root: str | PathLike[str])[source]#

Bases: object

The orchestrator’s root directory, holding one folder per job.

done() → Iterator[JobFolder][source]#: Iterate over the job folders that completed successfully.

folder_for(key: Hash | str) → JobFolder[source]#: The folder a job key maps to, whether or not it exists.

metric_names() → list[str][source]#: Every metric name recorded by a completed job, sorted.

static name_for(key: Hash | str) → str[source]#: The folder name for a job key.

promote(writer: JobWriter) → JobFolder[source]#

Move a staged job to its final path, replacing any earlier attempt.

The replace is a rename, so the final path never holds a partially written job. Removing an earlier attempt first is safe: we only get here having decided to re-run it, and a crash in between leaves no marker, which the next pass reads as “not done”.

prune_staging() → int[source]#

Remove leftover staging directories from crashed jobs.

Returns:: The number of directories removed.

property root: Path#: The root directory.

staging_path(key: Hash | str) → Path[source]#: The staging directory a job is built in, beside the job folders.

writer(key: Hash | str, manifest: Mapping[str, Any]) → JobWriter[source]#: Start building a job.

class krum.orchestration.storage.JobWriter(store: JobStore, key: str, manifest: Mapping[str, Any])[source]#

Bases: MetricRecorder

Builds one job in a staging directory, then promotes it atomically.

adopt(metrics: Mapping[str, dict[str, Any]]) → None[source]#: Take on the metrics a child process registered on our behalf.

discard() → None[source]#: Remove the staging directory without promoting it.

finish(status: str, error: str | None = None) → JobFolder[source]#

Write the manifest, the marker and the fingerprint, then promote.

Parameters:

status – Either done or failed.
error – The traceback to record in the FAILED marker.

Returns:

The promoted job folder, at its final path.

property key: str#: The job key.

record_called(called: Mapping[str, str] | None) → None[source]#: Record the functions the job entered, when it was traced.

class krum.orchestration.storage.MetricRecorder(path: str | PathLike[str], create: bool = False)[source]#

Bases: object

Where a running job writes its metrics: a directory and its open sinks.

This is the half of a job that a child process can own. It knows the directory being built and the metrics registered in it, but nothing about promoting that directory, which stays with the parent.

Metrics are flushed as they are pushed, so a crashed job’s partial output stays readable where it was staged, without ever being promoted to the job’s final path where the read path would pick it up.

close() → dict[str, dict[str, Any]][source]#

Close every open sink and return what was registered.

A sink that counted its rows contributes that count, so the size of a metric is recorded rather than needing the file read back to learn it.

A child process returns this to its parent, which is how a job run out of process still ends up with its metrics named in the manifest.

metric_path(name: str, dtype: str) → Path[source]#: Register a metric and return the file it writes to.

property metrics: dict[str, dict[str, Any]]#: The metrics registered so far, by name.

property path: Path#: The directory this job is being built in.

property sinks: dict[str, Any]#

The open metric writers, by metric name.

Held here rather than on the metric objects so that constructing the same metric twice inside one job keeps appending to one file.

krum.orchestration.storage.abbreviate(value: Any, length: int = 12) → str[source]#: Shorten a value for a human-readable reason, hashes especially.

krum.orchestration.storage.bind_job(recorder: MetricRecorder) → Iterator[MetricRecorder][source]#: Make a job current for the duration of a block.

Assemble the manifest recorded alongside a job’s output.

Parameters:

key – The job key.
callable – The user-defined function the job executes.
params – The bound parameters, in signature order.
owned – The module prefixes the key was computed over.
lock – The lock file to fingerprint.
start – A directory inside the repository to read provenance from, and to discover the lock file from.

Returns:

The manifest, ready to be written as JSON.

krum.orchestration.storage.current_job() → MetricRecorder | None[source]#: The job being executed in this context, if any.

krum.orchestration.storage.display_value(encoded: Any) → Any[source]#

Render an encoded parameter as a single cell value.

Parameters:: encoded – A value as returned by encode_value().
Returns:: The scalar itself, a tagged value’s short name, or a compact string.

krum.orchestration.storage.drift(recorded: Mapping[str, Any], current: Mapping[str, Any]) → list[str][source]#

Explain why a stored result no longer matches the current environment.

Parameters:

recorded – The witnesses stored with a completed job.
current – The witnesses of the environment now.

Returns:

One reason per witness that changed, empty when the result still holds.
A job with no recorded witnesses cannot be checked, which counts as a
reason – an unverifiable result is treated as stale, erring towards a
needless re-run rather than a stale answer.

krum.orchestration.storage.encode_params(params: Mapping[str, Any]) → dict[str, Any][source]#: Encode a parameter mapping, preserving its order.

krum.orchestration.storage.encode_value(value: Any) → Any[source]#

Encode a parameter value as JSON, keeping enough to trace it back.

Scalars pass through. A class or function becomes a tagged object carrying both its full location, for traceability, and its short name, which is what reads well as a column value.

Parameters:: value – The parameter value to encode.
Returns:: A JSON-serializable encoding of the value.

krum.orchestration.storage.environment_fingerprint(lock: str | PathLike[str] | None = None, start: str | PathLike[str] | None = None) → dict[str, Any][source]#

Describe the environment a job ran in.

The lock file hash stands in for the resolved dependency versions, which is exact and cheap next to sniffing each distribution. This lives in the fingerprint rather than in the job key, so that a dependency bump marks stored results stale without orphaning them under a new key.

Parameters:

lock – The lock file to hash; discovered from start when omitted.
start – Where to look for a lock file; the current directory by default.

Returns:

A JSON-serializable description of the environment. A lock file that
cannot be found, or that does not exist, is recorded as None rather
than raising – an environment without one is still worth describing.

krum.orchestration.storage.find_lock(start: str | PathLike[str] | None = None) → Path | None[source]#

Find the nearest uv.lock at or above a directory.

Parameters:: start – Where to start looking; the current directory by default.
Returns:: The path to `uv.lock`, or None if there is none above `start`.

krum.orchestration.storage.git_provenance(start: str | PathLike[str] | None = None) → dict[str, Any] | None[source]#

Describe the working tree a job ran from, so an output can be traced back.

A dirty tree is recorded rather than rejected: the orchestrator warns, and leaves it to the user to decide whether the result is publishable.

Parameters:: start – A directory inside the repository; the current one by default.
Returns:: The commit and whether the tree was dirty, or None outside a repository.

krum.orchestration.storage.hash_file(path: Path, cache: bool = True) → str[source]#

Hash a file’s contents, as a hex digest.

Parameters:

path – The file to hash.
cache – Reuse a digest computed earlier for the same path, size and modification time. A sweep fingerprints the same lock file once per job, which is worth not re-reading every time.

Returns:

The hex digest of the file’s contents.

krum.orchestration.storage.metric_filename(name: str) → str[source]#

Encode a metric name as a file name, reversibly.

Metric names are unrestricted, so they are percent-encoded rather than sanitized, keeping the mapping unambiguous in both directions.

krum.orchestration.storage.metric_name(filename: str) → str[source]#: Recover a metric name from its file name.

krum.orchestration.storage.read_manifest_params(manifest: Mapping[str, Any]) → dict[str, Any][source]#: The manifest’s parameters, rendered one cell value each.

krum.orchestration.storage.witnesses_of(environment: Mapping[str, Any]) → dict[str, Any][source]#

Reduce a recorded environment to the facts a stored result depends on.

These are compared on a later pass to decide whether a completed job is still valid. Deriving them from the recorded environment, rather than computing them separately, is what keeps deps.json from disagreeing with the manifest beside it.

The lock file hash is the one witness that is not already covered by the job key: third-party code is folded into a key as a location only, never by content, so a dependency bump is invisible to the key by design. The interpreter and the optimization flag are recorded too, being cheap and robust should key derivation ever stop depending on bytecode.

The interpreter is compared on major and minor only. Bytecode is stable across patch releases, so a patch bump is not a reason to discard results.

Parameters:: environment – An environment as recorded by environment_fingerprint().
Returns:: The witnesses, as comparable JSON values.