loom.pipeline is the pipeline description, planning, and stage orchestration layer
for loom.
It exists to turn a resolved experiment configuration into a validated, artifact-based execution graph. It should define pipeline and stage specs, validate dependencies, bind upstream artifacts to downstream inputs, create stage contexts, coordinate execution through executors, record stage state, and provide enough information for safe resume.
The pipeline layer should treat stages as generic Python components. A stage may
train a model, preprocess data, call a command-line tool, generate a report, or
do any other project-specific work. loom.pipeline should only care about the
contract around that work:
what stage is this?
what inputs does it consume?
what outputs does it promise?
what resources/runtime does it request?
what artifacts and status did it produce?
The initial design should be intentionally narrow. The first implementation
should support explicit static DAGs, local execution, artifact passing, status
tracking, and resume planning. It should avoid dynamic DAG mutation, nested task
scheduling, domain-specific stage types, and scheduler features that would turn
loom into a general workflow platform too early.
This document expands the pipeline DAG, stage execution, artifact passing, status
tracking, and resume responsibilities from loom.md. The package-wide
boundary remains strict: loom.pipeline owns orchestration around stages, while
project code owns stage internals, data semantics, metrics, training, reports,
and any domain-specific recovery.
Use this architecture:
Configuration:
compose, expand, validate, and instantiate configured objects
Pipeline:
validate static DAGs, bind artifacts, plan execution, run stages, track state
Project code:
implement concrete stage behavior
Stores:
persist artifacts, run state, logs, fingerprints, and indexes
Executors:
decide where and how individual stage invocations run
This means loom.pipeline is not an experiment framework, a data framework, or a
training framework. It is the generic runtime kernel for artifact-based research
pipelines.
The central pipeline boundary is:
loom.pipeline owns stage orchestration.
stage implementations own stage internals.
For example, loom.pipeline may decide whether to run the train stage based on
its fingerprint and input artifacts. It should not know how to resume a PyTorch
checkpoint, how to load a dataset, or how to compute a validation metric.
Owns shared primitives.
Responsibilities:
resource references
artifact references
records and manifests
fingerprint helpers
provenance value objects
shared exceptions
serialization helpers
Owns configuration composition and object construction.
Responsibilities:
load experiment configs
apply overlays
apply CLI overrides
expand recipes
resolve interpolation
validate config-level schemas
instantiate `_target_` objects
produce resolved config and config provenance data for the runner/run store
weave may produce a resolved pipeline config or instantiate a
PipelineSpec, but it should not execute stages.
Owns pipeline modeling and orchestration.
Responsibilities:
pipeline and stage specs
stage protocol
stage context
DAG validation
topological planning
artifact input binding
stage output validation
stage status transitions
runner lifecycle
resume decisions
executor integration
run selection
Owns persistence interfaces used by pipeline execution.
Responsibilities:
run directory creation
stage status files
input and output manifests
artifact indexes
fingerprint files
atomic writes
local artifact storage
local run storage
locking, later
Stores should be inspectable without Python. The local store should prefer plain
files such as status.json, inputs.json, outputs.json,
fingerprint.json, and artifacts.json.
Owns stage invocation mechanisms.
Responsibilities:
local in-process execution
post-v0 subprocess execution
post-v0 SLURM script generation and submission
executor result reporting
log capture integration
post-v0 runtime/resource mapping
Executors should not reinterpret pipeline semantics. They receive an execution request for a stage and return an execution result.
Owns command-line presentation only.
Responsibilities:
parse CLI arguments
call config and pipeline APIs
format plan/status/log/artifact output
return appropriate exit codes
The CLI should remain thin. It should not directly validate DAGs, write status files, or implement resume policy.
Owns concrete pipeline behavior.
Responsibilities:
stage implementations
domain-specific data loading
domain-specific artifact schemas
model training
metrics
plots
reports
project recipes
project codecs
Project code may depend on loom. loom must not depend on project code.
PipelineSpec
StageSpec
Stage protocol
StageContext
explicit static DAGs
unique stage names
declared stage inputs and outputs
DAG validation
topological ordering
branching and diamond DAGs
artifact input binding
stage output validation
local in-process executor
basic runner lifecycle
run directory creation through stores
stage status transitions
stage inputs/outputs files
stage fingerprints
resume planning
force/from/only/skip stage selectors
clear path-aware errors
small public Python API
The v0 target should be a fully working local synthetic pipeline:
write -> add -> multiply -> report
That pipeline should run without any domain package and should prove artifact passing, branching, status tracking, and resume behavior.
runtime DAG mutation
stage-internal subtask scheduling
general distributed scheduling
web dashboard
database-backed orchestration service
complex conditional expression language
automatic cloud execution abstraction
remote artifact stores
container orchestration
Bayesian sweeps
application-specific artifact schemas
application-specific stage base classes
application-specific stage registries
The first version should have one primary pipeline model:
an explicit static DAG of stages connected by named artifacts
Dynamic behavior can be added later only where it does not weaken planning, resume, provenance, and SLURM execution.
A static description of a pipeline DAG.
It defines the stages, their dependencies, their logical inputs and outputs, and pipeline-level defaults. It does not execute anything.
A static description of one stage in a pipeline.
It defines the stage name, target implementation, input bindings, output declarations, stage config, runtime/resource requests, and optional execution policy.
A concrete Python object with run(context, inputs) that performs work.
Stages are supplied by project code. They are checked structurally rather than through inheritance-heavy base classes. Plain callable stages are post-v0.
The import path that identifies the stage implementation.
In authored YAML, the stage target must be declared as:
factory:
_target_: project.stages.BuildManifestStagefactory.init is optional constructor kwargs and is distinct from config.
The runtime context passed to a stage invocation.
It gives the stage access to run paths, stage paths, resolved config, stage config, stores, logger, provenance helpers, selected runtime metadata, and output path allocation.
A stable name for a stage output within the pipeline graph.
Examples:
build_manifest.manifest
train.best_checkpoint
evaluate.metrics
Logical names are used for dependency binding, inspection commands, artifact indexes, and provenance.
The mapping from a stage input name to an upstream artifact.
Example:
inputs:
manifest: build_manifest.manifest
checkpoint: train.best_checkpointThe mapping from a stage output name to expected artifact metadata.
Example:
outputs:
metrics:
artifact_type: metrics.json
codec_key: json.v1A deterministic plan derived from a PipelineSpec, run store state, selected
stages, and resume policy.
It records which stages should run, skip, reuse, or be considered stale, and why.
The structured result of a stage invocation.
At minimum it should include returned output ArtifactRefs or a failure result
with exception/exit information.
The graph should be fully known before execution starts.
This enables:
early validation
dry-run planning
SLURM dependency generation
safe resume
artifact lineage
clear provenance
Dynamic DAG mutation should be out of scope until static DAG behavior is stable.
loom.pipeline should not inspect or control stage internals.
It should require only:
declared inputs
declared outputs
structural run interface
returned ArtifactRefs
status/failure reporting
The stage may perform arbitrary Python work internally.
Stages should communicate through explicit artifacts, not through implicit Python object sharing.
Good:
stage A writes an ArtifactRef
stage B receives that ArtifactRef as an input
Avoid:
stage A returns a live Python object consumed directly by stage B
Explicit artifacts make subprocess, SLURM, resume, and provenance behavior possible.
Running should always be based on an execution plan.
Even if the user calls a direct run() API, the runner should internally:
validate the spec
load prior run state
resolve stage selectors
bind inputs
compute fingerprints
decide run/reuse/skip/stale actions
then execute
This keeps dry-run behavior and real execution consistent.
A stage should be marked SUCCEEDED only after:
stage invocation completed
declared outputs were returned
artifact refs were validated
required artifact files exist when locally checkable
outputs were persisted
fingerprint was persisted
run artifact index was updated
Partial outputs should not be treated as valid.
Resume should not be based only on file existence.
The planner should compare:
stage target
stage config
input artifact refs/checksums/fingerprints
relevant resolved config subtree
selected environment/runtime metadata
loom version or pipeline contract version
optional user-provided fingerprint fields
If a stage has outputs but no matching fingerprint, it should not be reused by default.
Selectors such as from-stage, only-stage, force-stage, and skip-stage
should produce inspectable plans.
The same selectors should be available through the Python API and CLI.
Errors should include the best available location:
config path
stage name
input name
output name
artifact logical name
run directory path
status file path
The most common user mistakes should fail before execution starts.
Example:
pipeline:
stages:
- name: build_manifest
factory:
_target_: project.stages.BuildManifestStage
outputs:
manifest:
artifact_type: manifest
codec_key: json.v1
- name: summarize
factory:
_target_: project.stages.SummarizeStage
inputs:
manifest: build_manifest.manifest
outputs:
summary:
artifact_type: summary.json
codec_key: json.v1The authored config may be compact. After weave resolves it, the pipeline
layer should receive an explicit structure.
Example:
pipeline:
stages:
- name: train
factory:
_target_: project.stages.TrainStage
inputs:
manifest: build_manifest.manifest
outputs:
best_checkpoint:
artifact_type: checkpoint
- name: evaluate_main
factory:
_target_: project.stages.EvaluateStage
inputs:
checkpoint: train.best_checkpoint
outputs:
metrics:
artifact_type: metrics.json
codec_key: json.v1
- name: evaluate_external
factory:
_target_: project.stages.EvaluateStage
inputs:
checkpoint: train.best_checkpoint
outputs:
metrics:
artifact_type: metrics.json
codec_key: json.v1One upstream artifact can be consumed by multiple downstream stages. This should fall naturally out of DAG binding.
Example:
prepare
-> train_a
-> train_b
train_a.metrics + train_b.metrics
-> compare
The compare stage should not run until both upstream metrics artifacts are
available and valid.
Most dependencies should be expressed through input bindings. Occasionally a stage may need to wait for another stage even without consuming its artifacts.
Support:
depends_on:
- setup_environmentUse this sparingly. Artifact inputs are preferred because they carry stronger lineage and resume information.
Stage-specific configuration should live under config:
- name: train
factory:
_target_: project.stages.TrainStage
inputs:
manifest: build_manifest.manifest
outputs:
best_checkpoint:
artifact_type: checkpoint
config:
max_epochs: 100
batch_size: 32factory.init carries constructor kwargs and is separate from stage runtime
config.
Recommended rule:
factory.init defines stage-object constructor arguments
stage config defines the particular invocation
In simple cases these may overlap, but the pipeline layer should keep the distinction available.
Repeated stage patterns can be useful, but they should still produce an explicit static DAG before execution starts.
Examples:
evaluate one checkpoint across several datasets
generate one report per cohort
validate several model variants with the same stage target
Prefer config-time or recipe-time expansion for this. loom.pipeline should
receive the expanded PipelineSpec, validate it normally, and avoid runtime DAG
mutation.
from dataclasses import dataclass, field
from typing import Any, Mapping, Sequence
@dataclass(frozen=True, slots=True)
class PipelineSpec:
stages: Sequence[StageSpec]
name: str | None = None
description: str | None = None
metadata: Mapping[str, Any] = field(default_factory=dict)
schema_version: int | None = NoneSuggested meanings:
name:
human-readable pipeline name
description:
optional documentation string
stages:
ordered or unordered collection of StageSpec objects
defaults:
deferred in v0; authored defaults should fail validation clearly
metadata:
generic user/project metadata, not interpreted by loom unless documented
schema_version:
optional pipeline spec schema version; omitted means no schema version
constraint
Authored stage order should not be used as a substitute for dependencies.
Recommended behavior:
stages may be authored in any order
dependencies determine execution order
stable topological sorting preserves authored order where possible
This makes diffs readable while keeping semantics explicit.
Pipeline-level defaults are post-v0. Authored pipeline.defaults should fail
validation clearly in v0.
Pipeline defaults may provide values for stages that omit them.
Example:
pipeline:
defaults:
runtime:
executor: local
retry:
max_attempts: 1
stages:
- name: prepare
factory:
_target_: project.stages.PrepareStageDefault application should be deterministic and shallow enough to explain:
stage-level values override pipeline defaults
mapping defaults may merge recursively
lists should replace, not splice
Keep this aligned with config merge semantics where possible.
from dataclasses import dataclass, field
from typing import Any, Mapping, Sequence
@dataclass(frozen=True, slots=True)
class StageSpec:
name: str
factory: StageFactorySpec
inputs: Mapping[str, str] = field(default_factory=dict)
outputs: Mapping[str, "OutputSpec"] = field(default_factory=dict)
dependencies: Sequence[str] = field(default_factory=tuple)
stage_config: Mapping[str, Any] = field(default_factory=dict)
fingerprint_fields: Mapping[str, Any] = field(default_factory=dict)
resources: Mapping[str, Any] = field(default_factory=dict)Authored YAML keys are parsed into these internal fields:
factory -> StageFactorySpec
config -> stage_config
depends_on -> dependencies
fingerprint -> fingerprint_fields
Suggested meanings:
name:
stable stage identifier within the pipeline
factory:
includes constructor target path and optional `init` kwargs in
`factory.init`
inputs:
map of stage input name to upstream logical artifact name
outputs:
map of stage output name to declared output spec
dependencies:
explicit stage dependencies not represented by artifact inputs
stage_config:
invocation-specific config visible to the stage
fingerprint:
explicit stage-level semantic fingerprint fields
resources:
typed resource entries validated by ResourceRequest; Loom does not interpret
scheduler fields or include resources in semantic fingerprints by default
runtime:
deferred in v0; authored `runtime` fields should fail validation clearly
retry:
deferred in v0; authored `retry` fields should fail validation clearly
when:
deferred in v0; authored `when` fields should fail validation clearly
metadata:
deferred at the stage-spec level in v0; authored `metadata` fields should
fail validation clearly
V0 intentionally rejects unknown stage-level orchestration keys instead of
preserving them silently. runtime, retry, when, and stage-level
metadata are post-v0 fields.
Stage names should be:
unique within a pipeline
stable across reruns
usable as path components after normalization
clear in logs and status output
Recommended validation:
not empty
no slash
no backslash
no control characters
no "." or ".."
unique after normalization if normalization is applied
Avoid generating opaque stage names in core. Recipes may generate names, but the resolved config should show them explicitly.
The compact input syntax should be:
inputs:
checkpoint: train.best_checkpointThis means:
this stage receives input name "checkpoint"
from stage "train"
output "best_checkpoint"
An expanded form can be supported later if needed:
inputs:
checkpoint:
stage: train
output: best_checkpoint
artifact_type: checkpointThe compact string should be enough for v0.
V0 fields:
@dataclass(frozen=True, slots=True)
class OutputSpec:
artifact_type: str
codec_key: str | None = None
schema_version: int | None = None
metadata: Mapping[str, Any] = field(default_factory=dict)Suggested meanings:
artifact_type:
generic type label used for validation and inspection
path:
deferred in v0; physical paths are allocated by the artifact store
codec_key:
optional codec expected for the artifact
schema_version:
optional artifact schema version; omitted means no schema version constraint
required:
deferred in v0; declared outputs are required
metadata:
generic metadata
For v0, declared outputs are required, and authored output path fields should
fail validation clearly. Optional outputs and path templates can be added when
there is a concrete use case.
The pipeline layer stores authored resources as frozen plain data and validates
them through ResourceRequest. They have no executor-specific semantics, no
scheduler mapping, and no default impact on stage fingerprints.
Example:
resources:
entries:
cpu:
kind: cpu
amount: 16
unit: count
attributes: {}
memory:
kind: memory
amount: 64
unit: GiB
attributes: {}
gpu:
kind: gpu
amount: 1
unit: count
attributes: {}Validation checks:
resource kind syntax
entry mapping keys match entry kinds
built-in amount, unit, and attribute rules
unregistered resource kinds are rejected unless validation receives an
explicitly composed registry
Runtime settings are post-v0. V0 should reject authored stage runtime fields
with a clear validation error.
Runtime settings describe how the stage should be invoked.
Example:
runtime:
executor: local
container: null
environment:
OMP_NUM_THREADS: "8"The pipeline layer should not implement or preserve runtime settings in v0.
Authored stage runtime fields must fail validation clearly. Preserving runtime
settings for executors and fingerprints is post-v0.
Use a structural protocol:
from typing import Protocol
class Stage(Protocol):
def run(
self,
context: "StageContext",
inputs: Mapping[str, ArtifactRef],
) -> Mapping[str, ArtifactRef]:
...The exact signature should be stable before public release. The important design choice is that stages receive:
context:
runtime services and paths
inputs:
bound ArtifactRefs from upstream stages
and return:
mapping of declared output name to ArtifactRef
Optionally support plain callables later:
def summarize(context: StageContext, inputs: Mapping[str, ArtifactRef]) -> dict[str, ArtifactRef]:
...For v0, supporting objects with run() is sufficient. Plain callables can be
adapted after the object protocol is stable.
Pipeline stage mappings are orchestration specs, not generic _target_ object
graphs.
For v0, pipeline imports the stage factory target from
StageSpec.factory.target_path, constructs with StageSpec.factory.init, and
then invokes:
stage.run(context, inputs)
Authored config is parsed into StageSpec.stage_config and exposed through
StageContext; it is not copied into constructor kwargs. Accepting already
instantiated stage objects is supported only when factory.init is empty in this
phase, and callable/class constructor behavior is explicit.
Stages should be documented as expected to be idempotent with respect to their declared inputs and config.
Recommended behavior inside stage implementations:
write temporary outputs
validate outputs
return ArtifactRefs only for completed outputs
avoid mutating upstream artifacts
avoid relying on global hidden state
loom.pipeline cannot guarantee arbitrary stage idempotence, but its lifecycle
should make the safe path straightforward.
Pipeline resume and stage-internal resume are separate.
Example:
loom pipeline resume:
decides whether to skip or rerun the train stage
training stage resume:
decides whether to load a checkpoint inside the train stage
loom.pipeline should allow a stage to inspect its stage directory and config,
but it should not implement domain-specific checkpoint recovery.
StageContext should remain a narrow stage-author API and not leak direct mutable
store handles.
import os
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any, Mapping
@dataclass(frozen=True, slots=True)
class StageContext:
run_uri: str
stage_name: str
resolved_config: Mapping[str, Any]
stage_config: Mapping[str, Any]
provenance: Mapping[str, Any] = field(default_factory=dict)
metadata: Mapping[str, Any] = field(default_factory=dict)Public context attributes should not include run_dir, stage_dir, run_store,
or artifact_store.
The facade helper set is:
def input_artifact(name: str) -> ArtifactRef: ...
def load_input(name: str, *, expected_type: str | None = None, codec_key: str | None = None) -> object: ...
def load_artifact(ref: ArtifactRef, *, expected_type: str | None = None, codec_key: str | None = None) -> object: ...
def save_artifact(name: str, obj: object, *, artifact_type: str, codec_key: str, schema_version: int = 1, metadata: Mapping[str, Any] | None = None, fingerprint: str | None = None) -> ArtifactRef: ...
def register_artifact(name: str, uri: str, *, artifact_type: str, codec_key: str | None = None, schema_version: int = 1, metadata: Mapping[str, Any] | None = None, fingerprint: str | None = None, checksum: str | None = None, allow_external: bool = False) -> ArtifactRef: ...
def register_local_artifact(name: str, path: str | os.PathLike[str], *, artifact_type: str, codec_key: str | None = None, schema_version: int = 1, metadata: Mapping[str, Any] | None = None, fingerprint: str | None = None, checksum: str | None = None, allow_external: bool = False) -> ArtifactRef: ...
def local_output_path(name: str, *, suffix: str = "") -> Path: ...
def local_workspace_path(*parts: str) -> Path: ...save_artifact() and register_artifact() validate declared output type,
codec key, and schema version before writing or registering.
local_output_path() validates the output name and suffix, creates parents, and
raises PipelineValidationError when local output helpers are unavailable.
local_workspace_path() should validate all path parts and create the workspace
directory on demand; it should raise PipelineValidationError when local access
is not available.
load_input() fails early for unknown input names and load_artifact() or
load_input() fail clearly when no artifact store is available.
Each StageSpec is one DAG node.
Node identity is the stage name.
Edges come from:
artifact input bindings
explicit depends_on entries
Artifact input bindings are stronger because they identify the exact upstream artifact required.
Use this compact syntax:
STAGE_NAME.OUTPUT_NAME
Recommended parser behavior:
split on the first dot or last dot consistently
reject missing stage or output parts
reject references to unknown stages
reject references to unknown outputs
Since stage and output names should not contain dots in v0, splitting on one dot is simplest. If dots in stage names are later allowed, the reference syntax should be revisited before release.
Topological sorting should be deterministic.
Recommended behavior:
preserve authored order among otherwise-ready stages
raise clear cycle errors
include the cycle path when possible
Multiple downstream stages may consume the same upstream output.
This should require no special stage behavior.
One downstream stage may consume artifacts from multiple upstream stages.
The planner should ensure all required upstream stages are successful or reused before the downstream stage runs.
depends_on creates an ordering edge but not an artifact binding.
Use cases:
setup stage that prepares a shared environment
barrier stage
manual dependency on an external side effect
These should be allowed but not encouraged for ordinary data flow.
Validate before planning or execution.
Required checks:
pipeline has at least one stage
stage names are unique
stage names are valid
stage targets are present
input bindings are syntactically valid
input bindings reference existing stages
input bindings reference declared upstream outputs
depends_on references existing stages
no cycles
output names are unique within a stage
output specs have artifact_type
runtime/resource specs are mappings when present
Before running a stage:
stage object has a usable run method
all required input artifacts are available
stage directory can be created
artifact directory can be created
After running a stage:
result is a mapping
all required declared outputs are present
no undeclared outputs unless policy allows them
all returned values are ArtifactRefs
returned artifact types match OutputSpec when declared
returned codec keys match OutputSpec when declared
local artifact paths exist when checkable
Recommended v0 policy:
fail on undeclared outputs
Reason:
strict outputs improve provenance, artifact indexing, and resume safety
An allow_extra_outputs policy can be added later if repeated use justifies it.
Recommended v0 policy:
all declared outputs are required
Optional outputs add complexity to downstream binding and resume. Add them later only with clear semantics.
For local artifact paths:
relative output paths should stay inside the stage artifact directory
absolute output paths should require explicit external-artifact policy
parent directory traversal should be rejected by default
The planner should receive:
PipelineSpec
resolved config or relevant config view
run store state
artifact store state
resume policy
stage selectors
runtime defaults
The planner should return an ExecutionPlan.
Recommended structure:
@dataclass(frozen=True, slots=True)
class ExecutionPlan:
pipeline: PipelineSpec
stage_plans: tuple[StagePlan, ...]
selected_stages: frozenset[str]
reasons: Mapping[str, str]Each StagePlan should include:
stage name
action
reason
input bindings
expected outputs
fingerprint
upstream dependencies
downstream dependencies
resources metadata
Recommended action vocabulary:
RUN
REUSE
SKIP
STALE
BLOCKED
Meanings:
RUN:
invoke the stage
REUSE:
prior successful outputs are valid and should be used
SKIP:
user selector or condition excludes this stage
STALE:
prior outputs exist but cannot be reused
BLOCKED:
cannot run because an upstream stage is failed, missing, or skipped
Execution may turn STALE into RUN when rerun is allowed. Keeping STALE in
the plan is useful for explanation.
Default policy:
reuse prior succeeded stage only if fingerprint matches, required artifacts
exist, and present checksums validate when the store can read the URI
rerun failed, incomplete, stale, or missing stages
rerun downstream stages if upstream artifact identity changed
fail if required upstream artifacts are missing and cannot be produced
Strict mode can later verify checksums before reuse.
Support selectors:
from_stage:
run this stage and all downstream stages
only_stages:
run exactly this stage, requiring inputs to already exist or be reusable
force_stages:
rerun this stage even if reusable; downstream invalidation applies
skip_stages:
exclude this stage and stages that require its outputs, unless their inputs are already satisfied
Selectors should be represented as structured Python values, not raw CLI strings.
loom plan should use the same planner as loom run.
Example output:
build_manifest REUSE fingerprint match
train RUN forced by selector
evaluate RUN upstream train changed
report BLOCKED missing input evaluate.metrics
Recommended flow:
1. Validate pipeline spec.
2. Create or open run directory.
3. Persist resolved config if not already persisted.
4. Load prior run state.
5. Build execution plan.
6. Persist execution plan.
7. For each runnable stage in dependency order:
a. mark stage RUNNING
b. write inputs.json
c. write fingerprint candidate
d. invoke executor
e. validate outputs
f. persist outputs.json
g. update artifact index
h. mark stage SUCCEEDED
8. On failure:
a. persist failure metadata
b. mark stage FAILED
c. stop or continue according to policy
9. Finalize run status.
Recommended status vocabulary:
PENDING
RUNNING
SUCCEEDED
FAILED
SKIPPED
STALE
CANCELLED
Meanings:
PENDING:
known but not started
RUNNING:
currently executing or was interrupted while executing
SUCCEEDED:
completed and outputs were validated
FAILED:
completed unsuccessfully with failure metadata
SKIPPED:
excluded by selector or condition
STALE:
previous result exists but is not reusable
CANCELLED:
explicitly stopped before completion
On opening an existing run directory, the planner should treat old RUNNING
stages conservatively.
Recommended behavior:
if RUNNING has no live owner or cannot be verified, treat as incomplete
do not reuse outputs without matching succeeded status and fingerprint
preserve old failure/interruption metadata for debugging
Default behavior:
stop pipeline on first failed stage
mark downstream stages BLOCKED or leave them PENDING in the plan
write failure metadata and log paths
return non-zero CLI exit code
Continue-on-failure can be added later for independent branches.
The runner and stores should use atomic write helpers for stage state files:
write temp file in same directory
fsync where appropriate
rename into place
Stage artifact atomicity is harder because stages own their internals, but the context should make temporary output paths easy to use.
Recommended protocol:
class Executor(Protocol):
def execute(self, request: "StageExecutionRequest") -> "StageExecutionResult":
...The request should include:
run_uri
stage spec
stage context
bound inputs
fingerprint
resources metadata
The result should include:
success flag
returned outputs, for in-process execution
exit code, for post-v0 subprocess execution
exception metadata
stdout/stderr log paths
executor metadata
Runs the stage in the current Python process.
Use cases:
unit tests
local development
small pipelines
CI smoke tests
The local executor may return ArtifactRefs directly from the stage call.
Post-v0.
Runs one stage through:
loom stage run --run-uri RUN_URI --stage STAGE_NAMEUse cases:
stage isolation
independent logs
closer to parent-managed worker execution
debugging stage entry points
Future subprocess executors should not require Python object sharing across
stages. They should communicate through run store files and artifact refs.
Submitted afterok backends use the separate loom stage-job run command when
there is no live parent process to finalize the stage.
Post-v0.
SLURM behavior should live in executor implementations, not in the core planner.
Initial SLURM modes:
single-job:
one dry-run script plans whole-run continuation in v6
afterok:
one dry-run script per RUN stage, with logical afterok dependencies in v6
Detailed SLURM behavior should have a separate design document. The pipeline layer should expose enough structured plan information for SLURM executors to generate scripts and dependencies. Live scheduler submission and job IDs are v7/later behavior.
Pipeline execution depends on store interfaces, but the detailed run-store design should live in a separate document.
The pipeline layer should assume a local v0 layout compatible with caller-provided runtime config snapshots:
runs/RUN_ID/
config/
resolved.yaml
stages/
STAGE_NAME/
status.json
inputs.json
outputs.json
fingerprint.json
failure.json
logs/
artifacts/
STAGE_NAME/
...
artifacts.json
plan.json
run.json
Current v1 composed-config runs add artifact-safe config persistence without
making loom.pipeline import weave classes:
runs/RUN_ID/
config/
composition_manifest.json
recipe_manifest.json
run.json metadata.config_provenance
For composed configs, those files are plain-data records. They preserve authored
composition, recipe, provenance, and fingerprint facts, and they do not write
default config/resolved.yaml or config/resolved.redacted.yaml snapshots,
resolver outputs, or raw source bytes. Plain mapping config snapshots remain
caller-provided runtime data, not v1 composed-config artifacts.
Required pipeline interactions:
create run directory
create stage directory
read prior stage status
write stage status
write stage inputs
write stage outputs
write stage fingerprint
update artifact index
allocate artifact paths
capture logs
The pipeline runner should go through store APIs rather than hard-coding every path. The local implementation can still use the plain layout above.
Recommended stage fingerprint inputs:
stage name
stage target import path
stage constructor identity when available
stage config
declared inputs
input artifact refs
input artifact checksums/fingerprints when available
declared outputs
explicit future opt-in runtime/resource fields that affect outputs
relevant code provenance
loom version or pipeline contract version
user-provided extra fingerprint fields
V0 excludes StageSpec.resources from semantic fingerprints by default.
Avoid noisy values:
wall-clock timestamp
run directory
random run ID
log path
temporary directory path
hostname, unless outputs depend on it
Each stage should persist:
fingerprint value
fingerprint algorithm/version
fingerprint input summary
created_at timestamp
loom version
The summary helps users understand why a stage reran.
Checksums and fingerprints are distinct:
checksum:
identity of stored bytes
fingerprint:
identity of the production recipe
Resume should prefer fingerprints for semantic reuse and checksums for corruption or existence validation.
If an upstream stage reruns and produces a changed artifact identity, downstream stages should be considered stale unless their fingerprints still match under the documented fingerprint policy.
The planner should explain this:
evaluate RUN upstream artifact changed: train.best_checkpoint
Conditional stages are useful, but v0 should avoid a general expression language.
Recommended v0 policy:
do not implement `when` initially
Or, if needed, support only:
when: analysis.enabledWhere analysis.enabled is a config path that resolves to a boolean.
Do not support:
arbitrary Python expressions
Jinja expressions
shell expressions
complex boolean language
Conditions complicate planning, blocked downstream stages, and provenance. Keep them limited.
Recommended API:
from loom.pipeline import (
PipelineSpec,
StageSpec,
OutputSpec,
StageContext,
PipelineError,
validate_pipeline,
plan_pipeline,
)
from loom.pipeline.execution import PipelineRunner, create_authority_backed_serial_run_store
from loom.pipeline.executors import LocalExecutor
from loom.pipeline.stores import LocalArtifactStore, LocalRunArtifactStorefrom loom.pipeline import validate_pipeline
validate_pipeline(spec)Should raise a path-aware PipelineValidationError or return a structured
validation result. Raising is simpler for v0.
from loom.pipeline import plan_pipeline
plan = plan_pipeline(
spec,
run_store=run_store,
artifact_store=artifact_store,
resume=True,
selectors=selectors,
)Planning should not invoke stages.
from loom.pipeline.execution import PipelineRunner
runner = PipelineRunner(
run_store=run_store,
artifact_store=artifact_store,
executor=LocalExecutor(),
)
result = runner.run(spec, resolved_config=cfg)The runner should return a structured result:
run_uri
final status
stage results
artifact index
plan
Functional CLI integration is future roadmap work. V1-post is Python-API-only:
it provides public Python pipeline/config APIs and no functional loom CLI
commands or console script entry points. When implemented, the pipeline package
should support CLI commands without becoming CLI-specific.
Future CLI commands can call Python APIs:
loom validate experiment.yaml
loom plan experiment.yaml
loom run experiment.yaml
loom stage run --run-uri RUN_URI --stage STAGE
loom status RUN_DIR
loom logs RUN_DIR STAGE
loom artifacts list RUN_DIR
Should:
compose config
expand recipes
resolve interpolation
build PipelineSpec
validate pipeline DAG and stage specs
not run stages
Should:
compose config
build PipelineSpec
open or create run planning context
compute execution plan
print stage actions and reasons
not run stages
Should:
compose config
persist resolved config
build PipelineSpec
plan execution
run selected stages
write run status
return non-zero on failure
Post-v0.
Should:
load run directory
load resolved config and pipeline spec
bind inputs for one stage
run exactly that stage
write status/outputs/failure metadata
This command is required for post-v0 parent-managed subprocess execution.
Submitted afterok execution uses loom stage-job run so the stage can finalize
itself from durable run-store state without a parent runner.
Recommended hierarchy:
class PipelineError(LoomError): ...
class PipelineValidationError(PipelineError): ...
class PipelinePlanningError(PipelineError): ...
class StageExecutionError(PipelineError): ...
class StageContractError(PipelineError): ...
class ArtifactBindingError(PipelineError): ...
class ResumeError(PipelineError): ...Invalid pipeline input binding.
Stage:
summarize
Input:
manifest
Reference:
build_manifest.manifest
Reason:
stage "build_manifest" does not declare output "manifest"
Pipeline DAG contains a cycle.
Cycle:
prepare -> train -> evaluate -> prepare
Stage returned an undeclared output.
Stage:
train
Output:
checkpoint_latest
Declared outputs:
best_checkpoint
Cannot reuse previous stage output.
Stage:
train
Reason:
fingerprint matched but required artifact is missing
Artifact:
train.best_checkpoint
Path:
runs/example/artifacts/train/best.ckpt
Test:
PipelineSpec construction
StageSpec construction
OutputSpec validation
stage name validation
input reference parsing
pipeline default application
Use synthetic pipeline shapes:
linear DAG
branching DAG
diamond DAG
unordered authored stages
cycle error
missing stage reference
missing output reference
depends_on-only edge
Test:
topological order
resume reuse
missing output rerun
fingerprint mismatch rerun
downstream invalidation
from-stage selector
only-stage selector
force-stage selector
skip-stage selector
blocked downstream stage
Use dummy stages:
WriteArtifactStage
ReadArtifactStage
FailingStage
MissingOutputStage
ExtraOutputStage
SleepStage
Test:
successful run
status transitions
inputs.json writes
outputs.json writes
artifact index updates
failure metadata writes
rerun after failure
reuse after success
interrupted RUNNING state handling
Test:
local executor calls stage.run
local executor captures exceptions
subprocess command construction
subprocess exit code handling
subprocess log path recording
SLURM script generation later without real submission
Provide reusable dummy stages and fixtures so downstream packages can validate
their own stages against the loom.pipeline contract.
Examples:
assert stage returns declared ArtifactRefs
assert stage writes outputs via context.local_output_path
assert stage can run with synthetic ArtifactRefs
Items 1-12 align with the v0 plan. Items 13-15 are post-v0 and must not be implemented during v0 phase work.
Build in this order:
PipelineSpec,StageSpec,OutputSpec, and validation helpers.- Logical artifact reference parser.
- DAG construction and deterministic topological sorting.
Stageprotocol andStageContext.- Local run and artifact store interfaces needed by the runner.
- Execution plan and stage action model.
- Fingerprint computation for stage plans.
- Local executor.
- Pipeline runner lifecycle with status writes.
- Stage output validation and artifact index updates.
- Resume/reuse planning.
- Stage selectors:
from_stage,only_stages,force_stages,skip_stages. - Subprocess stage command support, post-v0.
- CLI wrappers for validate, plan, run, and stage run, post-v0.
- SLURM executor support after subprocess mode is stable, post-v0.
Each step should include focused tests before the next step depends on it.
loom.pipeline should be the narrow orchestration layer inside loom.
It should support:
explicit static DAGs
stage specs
structural stage protocol
stage contexts
artifact-based input/output binding
deterministic planning
local execution
status tracking
fingerprint-based resume
stage selectors
path-aware errors
thin CLI integration
It should avoid:
domain-specific stage types
implicit Python object passing between stages
runtime DAG mutation
nested task scheduling
general distributed scheduler behavior
complex conditional expression languages
heavy runtime dependencies
This keeps loom.pipeline useful as a reliable research pipeline kernel while
leaving concrete scientific, modeling, and analysis behavior in project code.