molforge.reproducibility¶

reproducibility ¶

Emit a citable pipeline.yaml from a molforge output's provenance.

Most papers in this space don't ship reproducible code. molforge already records what produced an output — every engine wrapper attaches a :class:~molforge.core.provenance.Provenance (engine, version, parameters, inputs, and a pointer to the step it consumed) to result.metadata["provenance"]. This module turns that chain into a single, human-readable manifest — the artifact a methods section can point at:

from molforge.reproducibility import emit_pipeline

folded = esmfold.predict(sequence)
docked = vina.dock(folded, ligand)
emit_pipeline(docked, "pipeline.yaml")

The resulting file linearizes the provenance chain into ordered steps and adds a consolidated environment block (molforge / Python / platform versions and the engine versions that ran)::

molforge_pipeline: 1
generated: "2026-07-15T12:00:00+00:00"
environment:
  molforge_version: "0.6.0"
  python_version: "3.12.13"
  platform: "macOS-14.3-arm64"
  engines: {ESMFold: "1.0.3", Vina: "1.2.5"}
  engines_detected: {fpocket: "4.1"}   # only when a step recorded none
steps:
  - step: 1
    engine: ESMFold
    engine_version: "1.0.3"
    inputs: {sequence: "MKT..."}
    parameters: {num_recycles: 4}
  - step: 2
    engine: Vina
    ...
output: {type: DockingResult}

Engines that shell out to a native binary (fpocket, GROMACS, gnina) have no version to record when they run, so the environment block falls back to :func:molforge.engine_versions — the registry of every backend molforge can drive and what it finds installed. It is reported under its own engines_detected key because it is the weaker claim: what is installed when the manifest is written, which is not necessarily what ran. Call :func:engine_versions directly to capture the whole environment up front, before a long run, which is the sturdier habit.

The in-memory :class:PipelineManifest and its to_dict / to_json forms need no third-party dependency. Reading and writing the .yaml form needs PyYAML — an opt-in extra (pip install "molforge[repro]") so molforge's core stays numpy-only.

Replay¶

:func:replay re-executes a manifest's chain, threading each step's output into the next::

from molforge.reproducibility import load_pipeline, replay

manifest = load_pipeline("pipeline.yaml")
output = replay(manifest, context={"ligand": "aspirin.sdf"})

It resolves each step's engine from the registry (molforge's own wrappers plus anything under :mod:molforge.plugins), reconstructs the call with a per-operation replay handler (molforge ships predict / dock), and runs it. Handlers own the reconstruction, so the fragile "which recorded input is upstream vs. a literal" wiring is contained per operation rather than guessed globally.

Replay is inherently partial: engines must be installed, GPU steps need the hardware (replay orchestrates, it doesn't provide compute), and inputs that aren't literals (a docking receptor is really the previous step's output; a ligand may be a path that no longer exists) come from the previous step or a supplied context. An unresolvable input, an unknown engine, or an operation with no registered handler raises a clear :class:ReplayError. Register a handler for a custom operation with :func:register_replay_handler.

Aggregates¶

A :class:PipelineManifest describes one output. Plenty of results aren't one output: a consensus structure stands on several folds, a ranking on a screen of thousands. Emitting one manifest per contributing prediction answers the question in a form nobody can read, and restates the shared upstream work — the target preparation, the MSA — once per prediction.

:func:aggregate_manifest folds them into one :class:AggregateManifest. Every distinct computation appears once, keyed by :meth:~molforge.core.provenance.Provenance.content_id, carrying a count of how many outputs descend from it::

from molforge.reproducibility import aggregate_manifest

manifest = aggregate_manifest(ensemble.members)
print(manifest.describe())
# aggregate (5 outputs, 6 unique steps from 10, 4 deduplicated) - molforge 0.9.1
#   shared:
#     MMseqs2 -> 5 outputs  [6623ea6703b3]

Provenance itself stays linear — one parent pointer, so :meth:~molforge.core.provenance.Provenance.chain keeps meaning what it says and cache keys keep their shape. The many-to-one structure lives in the aggregate instead, which is where it belongs: it is a property of the question being asked of a set of outputs, not of any one of them.

:meth:PipelineManifest.aggregate does the same from manifests rather than live objects, so a directory of pipeline.yaml files off a cluster run folds together without the outputs still being in memory — which is the usual situation by the time anyone asks how a screen was produced. Both paths produce identical step ids, so the two are interchangeable.

BackendVersion `dataclass` ¶

BackendVersion(
    name: str,
    category: str,
    kind: str,
    requirement: str,
    available: bool,
    version: str = "",
    location: str = "",
    detail: str = "",
)

What molforge can determine about one engine or backend.

Attributes:

Name	Type	Description
`name`	`str`	molforge's name for the backend, matching the `engine` field a wrapper writes into :class:`Provenance` (`"Boltz"`, `"fpocket"`, `"GROMACS"`).
`category`	`str`	Which part of molforge drives it — `"folding"`, `"docking"`, `"pockets"`, `"md"`, `"freeenergy"`, `"generative"`, or `"runtime"` for the shared numeric and chemistry libraries whose version changes results everywhere.
`kind`	`str`	`"python"`, `"executable"`, or `"repo"`. See the module docstring — these are three different reasons a version can be missing.
`requirement`	`str`	What was looked for: the distribution name that resolved, or the executable name searched for on `$PATH`.
`available`	`bool`	Whether molforge can find this backend right now. Always `False` for `repo` backends, which molforge cannot detect without the `repo_dir` the caller supplies.
`version`	`str`	The detected version, or `""`. Empty is never an error — read :attr:`detail` for which of the several reasons applies.
`location`	`str`	Resolved path for an `executable` backend; `""` otherwise.
`detail`	`str`	Plain-language reason :attr:`version` is empty, or `""` when a version was found.

to_dict ¶

to_dict() -> dict[str, Any]

Convert to a JSON-serialisable plain dict.

Keys with nothing to say (location, detail) are omitted so an embedded manifest block stays readable.

PipelineStep `dataclass` ¶

PipelineStep(
    step: int,
    engine: str,
    operation: str = "",
    engine_version: str = "",
    timestamp: str = "",
    inputs: dict[str, Any] = dict(),
    parameters: dict[str, Any] = dict(),
)

One step of a pipeline — a single provenance entry, linearized.

Attributes:

Name	Type	Description
`step`	`int`	1-indexed position in the pipeline (1 = oldest / originating).
`engine`	`str`	Producer name (engine name or molforge function path).
`operation`	`str`	The engine method that produced the output (`"predict"` / `"dock"` / ...); `""` when unrecorded.
`engine_version`	`str`	Producer version, `""` when not exposed.
`timestamp`	`str`	ISO-8601 UTC time the step ran.
`inputs`	`dict[str, Any]`	Input identifiers (sequence, paths, hashes).
`parameters`	`dict[str, Any]`	Engine arguments that drove the step.

to_dict ¶

to_dict() -> dict[str, Any]

Convert to a plain, ordered dict for serialization.

from_dict `classmethod` ¶

from_dict(data: dict[str, Any]) -> PipelineStep

Rebuild from :meth:to_dict output; tolerant of missing keys.

PipelineManifest `dataclass` ¶

PipelineManifest(
    environment: dict[str, Any],
    steps: list[PipelineStep],
    generated: str = "",
    output: dict[str, Any] = dict(),
    schema_version: int = SCHEMA_VERSION,
)

A citable description of the workflow that produced an output.

Attributes:

Name	Type	Description
`environment`	`dict[str, Any]`	molforge / Python / platform versions, an `engines` map of the engine versions that ran, and — when a step's wrapper recorded none — an `engines_detected` map of what :func:`molforge.engine_versions` finds installed now.
`steps`	`list[PipelineStep]`	The pipeline steps, oldest-first.
`generated`	`str`	ISO-8601 UTC time the manifest was emitted.
`output`	`dict[str, Any]`	A short descriptor of the terminal output (`{"type": ..., "name": ...}`).
`schema_version`	`int`	The on-disk schema version.

describe ¶

describe() -> str

A compact human-readable summary, one line per step.

to_dict ¶

to_dict() -> dict[str, Any]

Convert to the on-disk dict shape (ordered, JSON/YAML-native).

from_dict `classmethod` ¶

from_dict(data: dict[str, Any]) -> PipelineManifest

Rebuild from :meth:to_dict output; tolerant of missing keys.

to_json ¶

to_json(*, indent: int | None = 2) -> str

Serialize to JSON text. No third-party dependency.

from_json `classmethod` ¶

from_json(text: str) -> PipelineManifest

Deserialize from JSON text.

to_yaml ¶

to_yaml() -> str

Serialize to YAML text. Requires the repro extra (PyYAML).

from_yaml `classmethod` ¶

from_yaml(text: str) -> PipelineManifest

Deserialize from YAML text. Requires the repro extra (PyYAML).

aggregate `classmethod` ¶

aggregate(
    manifests: Iterable[PipelineManifest],
) -> AggregateManifest

Fold several single-output manifests into one aggregate.

The counterpart to building an aggregate from live objects with :func:aggregate_manifest, for when the outputs are long gone and all you have is the manifests — a directory of pipeline.yaml files off a cluster run, say, which is the usual situation by the time someone asks how a screen was produced.

A manifest records no parent pointers, but its steps are ordered oldest-first, which says the same thing; the chain is rebuilt from that ordering, so a manifest read off disk deduplicates against a live object on exactly the same terms.

Parameters:

Name	Type	Description	Default
`manifests`	`Iterable[PipelineManifest]`	The manifests to fold together.	required

Returns:

Name	Type	Description
`An`	`AggregateManifest`	class:`AggregateManifest` in which every distinct step
	`AggregateManifest`	appears once.

Raises:

Type	Description
`ValueError`	If `manifests` is empty, or any of them has no steps.

Example

from pathlib import Path from molforge.reproducibility import PipelineManifest, load_pipeline runs = [load_pipeline(p) for p in Path("runs").glob("*.yaml")] # doctest: +SKIP combined = PipelineManifest.aggregate(runs) # doctest: +SKIP

AggregateStep `dataclass` ¶

AggregateStep(
    id: str,
    engine: str,
    operation: str = "",
    engine_version: str = "",
    inputs: dict[str, Any] = dict(),
    parameters: dict[str, Any] = dict(),
    parent: str | None = None,
    contributes_to: int = 1,
    first_run: str = "",
)

One distinct computation in an :class:AggregateManifest.

A step that a thousand predictions shared appears here once, with :attr:contributes_to recording that it fed a thousand of them.

There is no single timestamp, because a deduplicated step may have run many times; :attr:first_run keeps the earliest seen, which is what dates the work.

Attributes:

Name	Type	Description
`id`	`str`	Short, manifest-unique identifier — a truncated :meth:`~molforge.core.provenance.Provenance.content_id`.
`engine`	`str`	Producer name.
`operation`	`str`	The method that produced the output; `""` when unrecorded.
`engine_version`	`str`	Producer version; `""` when not exposed.
`inputs`	`dict[str, Any]`	Input identifiers.
`parameters`	`dict[str, Any]`	Arguments that drove the step.
`parent`	`str \| None`	:attr:`id` of the step this one consumed, or `None` for an originating step.
`contributes_to`	`int`	How many of the manifest's outputs descend from this step. `1` means it was unique to one output; higher means it was shared, and is the whole reason to aggregate.
`first_run`	`str`	Earliest ISO-8601 timestamp seen for this step.

to_dict ¶

to_dict() -> dict[str, Any]

Convert to a plain, ordered dict for serialization.

from_dict `classmethod` ¶

from_dict(data: dict[str, Any]) -> AggregateStep

Rebuild from :meth:to_dict output; tolerant of missing keys.

AggregateManifest `dataclass` ¶

AggregateManifest(
    environment: dict[str, Any],
    steps: list[AggregateStep],
    outputs: list[dict[str, Any]],
    generated: str = "",
    schema_version: int = AGGREGATE_SCHEMA_VERSION,
)

A citable description of the work behind many outputs at once.

:class:PipelineManifest answers "how was this one thing made". Plenty of results aren't one thing: a consensus structure stands on five folds, a ranking on a screen of thousands. Emitting one manifest per prediction answers the question in a form nobody can read, and restates the shared upstream work — the target preparation, the MSA — once per prediction.

This aggregates them. Every distinct computation appears once, keyed by :meth:~molforge.core.provenance.Provenance.content_id, with a count of how many outputs descend from it. A thousand predictions off one MSA yield one MSA step marked contributes_to: 1000, and the manifest is a page rather than a directory.

Attributes:

Name	Type	Description
`environment`	`dict[str, Any]`	molforge / Python / platform versions and engine versions, consolidated across every contributing output.
`steps`	`list[AggregateStep]`	The distinct steps, ordered so a step's :attr:`parent` always precedes it.
`outputs`	`list[dict[str, Any]]`	One descriptor per aggregated output — its type, name where it has one, and the `step` id it terminates at.
`generated`	`str`	ISO-8601 UTC time the manifest was emitted.
`schema_version`	`int`	The on-disk schema version.

n_outputs `property` ¶

n_outputs: int

How many outputs this manifest describes.

total_steps `property` ¶

total_steps: int

Chain positions before deduplication.

The sum of every output's chain length — what a directory of per-output manifests would have spelled out.

shared_steps `property` ¶

shared_steps: list[AggregateStep]

Steps more than one output descends from, most-shared first.

roots ¶

roots() -> list[AggregateStep]

The originating steps — those consuming nothing in this manifest.

step_by_id ¶

step_by_id(step_id: str) -> AggregateStep

Look a step up by :attr:AggregateStep.id.

Raises:

Type	Description
`KeyError`	If no step has that id.

summary ¶

summary() -> dict[str, int]

The headline counts, as emitted under the summary key.

describe ¶

describe() -> str

A compact human-readable summary.

Leads with the counts, then the shared steps — the ones worth looking at, since a step only many outputs depend on is the one whose settings matter most.

to_dict ¶

to_dict() -> dict[str, Any]

Convert to the on-disk dict shape (ordered, JSON/YAML-native).

from_dict `classmethod` ¶

from_dict(data: dict[str, Any]) -> AggregateManifest

Rebuild from :meth:to_dict output; tolerant of missing keys.

The summary block is not read back — it is derived from the steps, so trusting a stored copy would let a hand-edited file report counts its own steps contradict.

to_json ¶

to_json(*, indent: int | None = 2) -> str

Serialize to JSON text. No third-party dependency.

from_json `classmethod` ¶

from_json(text: str) -> AggregateManifest

Deserialize from JSON text.

to_yaml ¶

to_yaml() -> str

Serialize to YAML text. Requires the repro extra (PyYAML).

from_yaml `classmethod` ¶

from_yaml(text: str) -> AggregateManifest

Deserialize from YAML text. Requires the repro extra.

ReplayError ¶

Bases: RuntimeError

Raised when a manifest can't be replayed — a missing engine, an operation with no handler, or an input that can't be resolved.

engine_versions ¶

engine_versions(
    *,
    category: str | None = None,
    probe_executables: bool = True,
    timeout: float = DEFAULT_PROBE_TIMEOUT,
    refresh: bool = False,
) -> dict[str, BackendVersion]

Report every engine and backend molforge knows how to drive.

Parameters:

Name	Type	Description	Default
`category`	`str \| None`	Return only this category (`"folding"`, `"docking"`, `"pockets"`, `"md"`, `"freeenergy"`, `"generative"`, `"runtime"`). `None` returns all of them.	`None`
`probe_executables`	`bool`	Run each installed native binary's version flag. `False` skips every subprocess: executables are still reported as available or not via `$PATH`, but with an empty version. Use it when you want the sweep to cost nothing, or in a sandbox where spawning processes is unwelcome.	`True`
`timeout`	`float`	Seconds to wait for one probe. A tool that exceeds it is reported as available with no version, never as an error.	`DEFAULT_PROBE_TIMEOUT`
`refresh`	`bool`	Re-probe instead of reusing the memoized sweep. The result is cached because the answer only changes when something is installed, and a caller emitting one manifest per output would otherwise re-run every binary on the machine.	`False`

Returns:

Type	Description
`dict[str, BackendVersion]`	class:`BackendVersion` per backend, keyed by
`dict[str, BackendVersion]`	attr:`BackendVersion.name` and ordered by category. Every known
`dict[str, BackendVersion]`	backend appears, installed or not — an absent engine is a fact
`dict[str, BackendVersion]`	about the environment worth recording.

Raises:

Type	Description
`ValueError`	If `category` isn't one molforge uses. A typo would otherwise silently return nothing, which reads as "nothing is installed".

Example

from molforge import engine_versions md = engine_versions(category="md", probe_executables=False) sorted(md) ['AMBER', 'GROMACS', 'OpenMM', 'tleap']

pipeline_manifest ¶

pipeline_manifest(
    obj: Provenance | object,
) -> PipelineManifest

Build a :class:PipelineManifest from an output or a provenance.

Parameters:

Name	Type	Description	Default
`obj`	`Provenance \| object`	A :class:`~molforge.core.provenance.Provenance`, or any molforge output carrying one at `metadata["provenance"]` (a :class:`~molforge.core.Protein`, `DockingResult`, `Pose`, `DesignedSequence`, ...).	required

Returns:

Type	Description
`PipelineManifest`	A manifest with the provenance chain linearized oldest-first and the
`PipelineManifest`	environment consolidated.

Raises:

Type	Description
`ValueError`	If no provenance can be found on `obj`.

emit_pipeline ¶

emit_pipeline(
    obj: Provenance | object,
    path: str | PathLike[str],
    *,
    fmt: str = "yaml",
) -> PipelineManifest

Write a pipeline.yaml (or .json) describing how obj was made.

Parameters:

Name	Type	Description	Default
`obj`	`Provenance \| object`	An output carrying provenance, or a :class:`~molforge.core.provenance.Provenance`.	required
`path`	`str \| PathLike[str]`	Destination file path.	required
`fmt`	`str`	`"yaml"` (default; needs the `repro` extra) or `"json"` (no extra).	`'yaml'`

Returns:

Name	Type	Description
`The`	`PipelineManifest`	class:`PipelineManifest` that was written (handy for inspection).

Raises:

Type	Description
`ValueError`	If no provenance is found, or `fmt` is unrecognized.
`ImportError`	If `fmt="yaml"` and PyYAML isn't installed.

load_pipeline ¶

load_pipeline(
    path: str | PathLike[str],
) -> PipelineManifest

Load a manifest from a .yaml / .json file.

The format is chosen by suffix: .json is parsed as JSON (no extra); anything else is parsed as YAML (needs the repro extra). Since YAML is a superset of JSON, a .yaml loader also reads JSON content.

Raises:

Type	Description
`ImportError`	If a YAML file is loaded without PyYAML installed.

aggregate_manifest ¶

aggregate_manifest(
    objs: Iterable[Provenance | object],
) -> AggregateManifest

Build an :class:AggregateManifest from many outputs.

Parameters:

Name	Type	Description	Default
`objs`	`Iterable[Provenance \| object]`	Outputs carrying provenance (:class:`~molforge.core.Protein`, `DockingResult`, `Pose`, `DesignedSequence`, ...) or :class:`~molforge.core.provenance.Provenance` instances. Any mix is fine.	required

Returns:

Type	Description
`AggregateManifest`	A manifest in which every distinct computation appears once.

Raises:

Type	Description
`ValueError`	If `objs` is empty, or if any element carries no provenance.

Example

from molforge.reproducibility import aggregate_manifest manifest = aggregate_manifest(ensemble.members) # doctest: +SKIP print(manifest.describe()) # doctest: +SKIP aggregate (5 outputs, 6 unique steps from 10, 4 deduplicated) — molforge 0.9.1 shared: ESMFold v1.0.3 -> 5 outputs [a1b2c3d4e5f6]

emit_aggregate ¶

emit_aggregate(
    objs: Iterable[Provenance | object],
    path: str | PathLike[str],
    *,
    fmt: str = "yaml",
) -> AggregateManifest

Write an aggregate.yaml (or .json) spanning many outputs.

The many-output counterpart of :func:emit_pipeline.

Parameters:

Name	Type	Description	Default
`objs`	`Iterable[Provenance \| object]`	Outputs carrying provenance.	required
`path`	`str \| PathLike[str]`	Destination file path.	required
`fmt`	`str`	`"yaml"` (default; needs the `repro` extra) or `"json"`.	`'yaml'`

Returns:

Name	Type	Description
`The`	`AggregateManifest`	class:`AggregateManifest` that was written.

Raises:

Type	Description
`ValueError`	If `objs` is empty, provenance is missing, or `fmt` is unrecognized.
`ImportError`	If `fmt="yaml"` and PyYAML isn't installed.

load_aggregate ¶

load_aggregate(
    path: str | PathLike[str],
) -> AggregateManifest

Load an aggregate manifest from a .yaml / .json file.

Format is chosen by suffix, as in :func:load_pipeline.

Raises:

Type	Description
`ImportError`	If a YAML file is loaded without PyYAML installed.

register_replay_handler ¶

register_replay_handler(
    operation: str,
) -> Callable[[Callable[..., Any]], Callable[..., Any]]

Decorator registering a replay handler for operation.

A handler has the signature (engine_factory, step, upstream_output, context) -> output — it reconstructs the engine from step.parameters and calls the right method, using upstream_output (the previous step's result) and context (user-supplied inputs) as needed::

@register_replay_handler("dock")
def _dock(factory, step, upstream, context): ...

replay ¶

replay(
    source: PipelineManifest | Provenance | object,
    *,
    context: dict[str, Any] | None = None,
) -> Any

Re-execute a pipeline's chain, returning the terminal output.

Parameters:

Name	Type	Description	Default
`source`	`PipelineManifest \| Provenance \| object`	A :class:`PipelineManifest` (e.g. from :func:`load_pipeline`), a :class:`~molforge.core.provenance.Provenance`, or any output carrying one.	required
`context`	`dict[str, Any] \| None`	Objects to resolve recorded inputs that aren't literals — keyed by the input name (`{"ligand": "aspirin.sdf"}`). A step's primary input is threaded from the previous step automatically; `context` covers the rest.	`None`

Returns:

Type	Description
`Any`	The output of the final step.

Raises:

Type	Description
`ReplayError`	If a step has no recorded operation, its engine can't be resolved, no handler is registered for its operation, or a required input can't be resolved.

molforge.reproducibility¶

reproducibility ¶

Replay¶

Aggregates¶

BackendVersion dataclass ¶

to_dict ¶

PipelineStep dataclass ¶

to_dict ¶

from_dict classmethod ¶

PipelineManifest dataclass ¶

describe ¶

to_dict ¶

from_dict classmethod ¶

to_json ¶

from_json classmethod ¶

to_yaml ¶

from_yaml classmethod ¶

aggregate classmethod ¶

AggregateStep dataclass ¶

to_dict ¶

from_dict classmethod ¶

AggregateManifest dataclass ¶

n_outputs property ¶

total_steps property ¶

shared_steps property ¶

roots ¶

step_by_id ¶

summary ¶

describe ¶

to_dict ¶

from_dict classmethod ¶

to_json ¶

from_json classmethod ¶

to_yaml ¶

from_yaml classmethod ¶

ReplayError ¶

engine_versions ¶

pipeline_manifest ¶

emit_pipeline ¶

load_pipeline ¶

aggregate_manifest ¶

emit_aggregate ¶

load_aggregate ¶

register_replay_handler ¶

replay ¶

BackendVersion `dataclass` ¶

PipelineStep `dataclass` ¶

from_dict `classmethod` ¶

PipelineManifest `dataclass` ¶

from_dict `classmethod` ¶

from_json `classmethod` ¶

from_yaml `classmethod` ¶

aggregate `classmethod` ¶

AggregateStep `dataclass` ¶

from_dict `classmethod` ¶

AggregateManifest `dataclass` ¶

n_outputs `property` ¶

total_steps `property` ¶

shared_steps `property` ¶

from_dict `classmethod` ¶

from_json `classmethod` ¶

from_yaml `classmethod` ¶