---
myst:
  html_meta:
    description: Run and configure CyteArc RNA pipelines, inspect durable PipelineRun records, and access frozen analysis outputs.
---

# Analysis pipeline API reference

`DataStore.pipeline` runs CyteArc's standard RNA recipe and records every invocation in the datastore.
The result is a durable, read-only {py:class}`~cytearc.PipelineRun` whose outputs are exact immutable
{py:class}`~cytearc.ArtifactRef` values.

```{eval-rst}
.. autosummary::
   :nosignatures:

   cytearc.PipelineRun
   cytearc.PipelineExecutionError
   cytearc.datastore.pipeline_accessor.PipelineAccessor
   cytearc.datastore.pipeline_accessor.PipelineEvent
   cytearc.datastore.pipeline_accessor.PipelineAccessor.run
   cytearc.datastore.pipeline_accessor.PipelineAccessor.open
   cytearc.datastore.pipeline_accessor.PipelineAccessor.list_runs
   cytearc.datastore.pipeline_accessor.PipelineAccessor.abandon_label_claim
```

## Run the RNA recipe

```python
run = ds.pipeline.run(assay="RNA", label="baseline")

run["pca"]
run["cluster_selection"]
run["clusters"]
run.cells.to_pandas_dataframe(["umap_1", "umap_2", "clusters"])
ds.plots.embedding(run=run, layout="umap", color_by="clusters")
ds.markers.load(marker=run["markers"], group_id=0)
```

The public signature is:

```python
def run(
    *,
    assay: str | None = None,
    label: str | None = None,
    cell_key: str = "I",
    filtering: bool | Mapping[str, object] = True,
    harmony_batch_columns: Sequence[str] | None = None,
    hvg_count: int = 1000,
    pca_dims: int = 21,
    neighbors_k: int = 11,
    umap: bool = True,
    leiden: Mapping[str, object] | bool = True,
    cell_cycle: bool = True,
    paris: bool = True,
    doublets: bool = True,
    markers: bool = True,
    snapshot_columns: Sequence[str] = (),
    params: Mapping[str, object] | None = None,
    callback: PipelineCallback | None = None,
) -> PipelineRun: ...
```

The default recipe snapshots its inputs, filters cells, scores cell cycle, selects highly variable
genes, normalizes, runs PCA, builds ANN, neighbour, and connectivity artifacts, initializes UMAP,
runs UMAP, evaluates Leiden at `0.5`, `0.75`, `1.0`, and `1.25`, runs Paris, selects a clustering,
scores doublets, and searches for markers. Stages run one at a time in this fixed order on the
calling thread, so each stage keeps its own wall time, sampled resident memory, and artifact
receipt.

Harmony is enabled by a non-empty `harmony_batch_columns` sequence. The main graph then uses the
Harmony coordinates. When doublet scoring is enabled, its graph branch still uses the uncorrected
PCA coordinates. The branch artifacts appear in their stage receipt, not as convenience entries in
the run's top-level output mapping.

Boolean stage options disable that stage when set to `False`. `True` uses that stage's defaults.
A mapping configures the stage. `None` is rejected. An empty filtering mapping uses the same
automatic defaults as `filtering=True`. An empty Leiden mapping is invalid; a custom Leiden
request contains exactly `partitions`:

```python
run = ds.pipeline.run(leiden={"partitions": [0.4, 0.8]})
```

Doublets and markers require at least one Leiden candidate. Paris can still run as a diagnostic
output when those stages are disabled; a Paris-only run has `run["paris"]` and no
`run["clusters"]`. Setting `umap=False` skips UMAP, and the embedding initialization unless t-SNE,
which uses it, is enabled; a skipped stage's artifact does not appear in the completed run.

`filtering=True` uses MAD filtering over available assay QC columns. The default and
`method="auto"` both resolve to `method="mad"`. Set filtering to `False` to retain the captured
input selection, or pass a configuration mapping. MAD filtering accepts `attrs`, `n_mads`,
`min_cells_per_sample`, and optionally `sample_column`; without a sample column it uses pooled
bounds. Groups with fewer than 20 active cells are retained with a warning by default, including
small pooled selections. Pass `method="gaussian"` explicitly for the former Gaussian policy,
with `attrs`, `min_p`, and `max_p`. Gaussian filtering does not accept a sample column.
`method="manual"` requires aligned `attrs`, `lows`, and `highs`, with an
optional Boolean `keep_bounds` value. Probability, MAD, and manual-bound values must be finite
numbers when present; booleans and numeric strings are rejected rather than coerced.
If filtering is requested and none of the default QC columns exists, validation raises instead of
silently analyzing the unfiltered cells. Pass `filtering=False` only when that choice is deliberate.

```python
run = ds.pipeline.run(
    filtering={
        "method": "manual",
        "attrs": ["RNA_nCounts", "RNA_nFeatures"],
        "lows": [1000, 500],
        "highs": [15000, 4000],
    },
)
```

## Configure stages with one mapping

`params` configures the run with one mapping. Each key names a stage, and its value holds the
keyword arguments the pipeline forwards to that stage's function. The pipeline still supplies the
artifacts each stage consumes. A setting has exactly the meaning of the same keyword on the method
in the table below, and omitting it uses that method's default.

Settings are JSON values, so a run configuration can be saved and passed again; infinity and NaN are
never settings. Some settings spell a derived default as `None`: `hvg` `max_cells`, which excludes
genes detected in at least the number of selected cells minus 20; `batch_size` of every stage that
takes it; `cell_cycle` `ctrl_size` and `log_transform`; `paris` `min_cluster_size`; and
`normalization` `log_transform` and `renormalize_subset`, whose defaults follow the assay's
normalizer ({doc}`assays`). The run records `params` as given, and each artifact records the values
its method resolved.

```python
run = ds.pipeline.run(
    params={
        "filtering": {"method": "manual", "attrs": ["RNA_nCounts"], "lows": [500], "highs": [20000]},
        "hvg": {"min_mean": 0.01, "max_mean": 5.0, "keep_bounds": True},
        "pca": {"dims": 30},
        "neighbors": {"k": 15},
        "umap": {"n_epochs": 400, "min_dist": 1.0, "spread": 2.0, "parallel": True},
        "leiden": {"partitions": [0.6, 0.8, 1.0, 1.2, 1.4], "selected": 1.0},
        "membership_strength": True,
        "tsne": {"max_iter": 800, "early_iter": 200, "alpha": 10, "box_h": 0.7},
        "species": "homo_sapiens",
    },
)
```

| Section | Forwards to | Settings |
| --- | --- | --- |
| `filtering` | the `filtering` option | the filtering mapping above, or a bool |
| `cell_cycle` | cell-cycle scoring | `s_genes`, `g2m_genes`, `ctrl_size`, `log_transform`, `n_bins`, `rand_seed` |
| `hvg` | {py:meth}`~cytearc.datastore.namespaces.FeaturesAccessor.hvgs` | `top_n`, `min_cells`, `max_cells`, `min_mean`, `max_mean`, `min_var`, `max_var`, `n_bins`, `lowess_frac`, `blacklist`, `keep_bounds`, `bin_strategy` |
| `normalization` | {py:meth}`~cytearc.datastore.namespaces.FeaturesAccessor.normalize` | `log_transform`, `renormalize_subset` |
| `pca` | {py:meth}`~cytearc.datastore.namespaces.ReductionAccessor.pca` | `dims`, `feat_scaling`, `batch_size` |
| `harmony` | {py:meth}`~cytearc.datastore.namespaces.ReductionAccessor.harmony` | `batch_columns` (required), `harmony_params`, `batch_size` |
| `ann_index` | {py:meth}`~cytearc.datastore.namespaces.GraphAccessor.ann_index` | `ann_metric`, `ann_efc`, `ann_ef`, `ann_m`, `ann_parallel`, `rand_state`, `batch_size` |
| `neighbors` | {py:meth}`~cytearc.datastore.namespaces.GraphAccessor.neighbors` | `k`, `batch_size` |
| `connectivity` | {py:meth}`~cytearc.datastore.namespaces.GraphAccessor.connectivity` | `local_connectivity`, `bandwidth` |
| `embedding_initialization` | {py:meth}`~cytearc.datastore.namespaces.EmbeddingsAccessor.initialization` | `n_centroids`, `rand_state`, `batch_size`, `kmeans_sampling`, `kmeans_batch_size` |
| `umap` | {py:meth}`~cytearc.datastore.namespaces.EmbeddingsAccessor.umap` | `umap_dims`, `spread`, `min_dist`, `n_epochs`, `repulsion_strength`, `initial_alpha`, `negative_sample_rate`, `use_density_map`, `dens_lambda`, `dens_frac`, `dens_var_shift`, `random_seed`, `parallel`, `symmetric_graph`, `graph_upper_only` |
| `tsne` | {py:meth}`~cytearc.datastore.namespaces.EmbeddingsAccessor.tsne` | `tsne_dims`, `lambda_scale`, `max_iter`, `early_iter`, `alpha`, `box_h`, `symmetric_graph`, `graph_upper_only` |
| `leiden` | {py:meth}`~cytearc.datastore.namespaces.ClustersAccessor.leiden` | `partitions`, `selected`, `backend`, `random_seed`, `symmetric_graph`, `graph_upper_only` |
| `membership_strength` | {py:meth}`~cytearc.datastore.namespaces.ClustersAccessor.membership_strength` | none; a bool |
| `paris` | Paris clustering | `n_clusters`, `min_cluster_size` |
| `doublets` | {py:meth}`~cytearc.datastore.namespaces.QcAccessor.doublets` | `cluster_sample_fraction`, `max_cells_per_cluster`, `simulation_ratio`, `heterotypic_fraction`, `save_k`, `smoothing_t`, `normalize_scores`, `random_seed` |
| `markers` | marker search | none; a bool |
| `species` | recorded with the run | a species key such as `homo_sapiens` or `mus_musculus` |

Optional stages also accept `True` or `False`: `filtering`, `cell_cycle`, `harmony` (`False`
only), `umap`, `tsne`, `leiden`, `membership_strength`, `paris`, `doublets`, and `markers`. `tsne`
and `membership_strength` are off unless requested. t-SNE uses the UMAP graph and embedding
initialization and runs after UMAP; its coordinates appear as `tsne_1`, `tsne_2`, and so on.
A new t-SNE embedding needs the optional `sgtsnepi` package of the `tsne` extra; without it the
`tsne` stage fails, after the earlier stages have completed, and the run raises a
`PipelineExecutionError` whose `__cause__` is the `ImportError` that names the extra. A run whose
t-SNE embedding the store already holds reuses it without the package, so a run that enables t-SNE
without `sgtsnepi` is not refused: it logs a warning that names the extra before it creates a run
record, and then runs.
A saved configuration that sets `parallel` for t-SNE is rejected as an unknown setting when it
runs. An existing run record still reopens and reports, because reopening does not validate its
configuration against the current recipe.
Membership strength is computed on the saved clustering and appears as `membership_strength`: the
fraction of each cell's graph neighbors that carry the cell's own cluster label.

`pca` `dims=0` skips the `pca` stage. The graph is then built on the normalized values of the
selected features, one coordinate per feature: the ANN index, the neighbors, the embedding
initialization, and cluster selection use `run["normalized"]` as their coordinates, and the run
records no `pca` output. Harmony corrects reduced coordinates and doublet scoring needs a PCA graph,
so `dims=0` with `harmony_batch_columns`, or with doublet scoring on, which it is by default,
raises `ValueError` before a run record is created; pass `doublets=False`.
`leiden` `selected` names one of the partitions as the saved clustering, `run["clusters"]`, in place
of the silhouette choice, and the cluster-selection stage is skipped; the other partitions still
run beside it.

With `dims=0` the graph has one coordinate per highly variable gene, so the memory and time of the
ANN index, the neighbour queries, the embedding initialization, and cluster selection scale with
the HVG count instead of the PCA dimensions. The hnswlib index holds about
`4 * hvg_count + 8 * ann_m + 130` bytes per cell on x86-64 Linux, about 4.5 GB for 1,000,000 cells
with 1,000 HVGs and 8.5 GB with 2,000, and the `ann_index` and `neighbors` stages fail before they
create or load it when it does not fit the memory budget: the run raises a
`PipelineExecutionError` whose `__cause__` is the `MemoryError`. By default the embedding
initialization reads the coordinates in their stored row bands, and fits k-means in memory over
every cell's coordinates, 4 bytes per HVG per cell, when one band holds every cell; set
`params={"embedding_initialization": {"batch_size": ...}}` below the cell count to select the
streamed fit, which holds a sample of the cells and one block of coordinates at a time.

The `doublets` stage forces each simulated doublet, with probability `heterotypic_fraction` (0.8
by default), to pair parents from two different clusters, so it needs a saved clustering with two
or more clusters. The silhouette choice never saves a partition with one cluster, but `leiden`
`selected` can. The stage then fails with a `PipelineExecutionError` that names the settings to
change: select a partition with two or more clusters, set
`params["doublets"]["heterotypic_fraction"]` to 0 to simulate doublets from any two sampled cells,
or pass `doublets=False`.

Unknown sections or settings, settings without a JSON value such as infinity, and `True` for a stage
that always runs are rejected before a run record is created; `hvg` settings also get the checks of
`features.hvgs` then. `min_cells` must be an integer of at least 0, `top_n` and `n_bins` integers of
at least 1, `lowess_frac` a finite number from 0 to 1 for either `bin_strategy`, `keep_bounds` a
boolean, and `bin_strategy` `"adaptive"` or `"fixed"`; an omitted setting is checked as its
`features.hvgs` default.
`tsne` settings get the checks of `embeddings.tsne` then too, because the stage runs after most
others: `tsne_dims`, `max_iter`, and `alpha` must be integers of at least 1, `early_iter` an integer
of at least 0, `lambda_scale` and `box_h` finite positive numbers, and `symmetric_graph` and
`graph_upper_only` booleans; an omitted setting is checked as its `embeddings.tsne` default. The `umap`
`symmetric_graph` and `graph_upper_only` settings must be booleans too, checked at the same time;
`embeddings.umap` no longer accepts `None` for them. A setting
cannot also be given through its shortcut argument: `hvg_count` with
`hvg.top_n`, `pca_dims` with `pca.dims`, `neighbors_k` with `neighbors.k`, `harmony_batch_columns`
with `harmony`, or a changed stage switch such as `umap=False` with `params["umap"]`. Each shortcut
defaults to the default of the keyword it sets, so `hvg_count`, `pca_dims`, and `neighbors_k` follow
`features.hvgs` `top_n`, `reduction.pca` `dims`, and `graph.neighbors` `k`. Every other stage checks its own
values when it runs. The run's configuration records the settings under `params` as JSON values.

The invocation is validated before a run record is created. Unknown options, missing columns,
invalid stage combinations, reserved snapshot fields, and an already completed label fail without
starting a run. A handled stage failure raises {py:class}`~cytearc.PipelineExecutionError`. Use its
`run_id` to inspect the persisted failure, and its `stage` to identify the failing stage. The
original exception is retained as `__cause__`.

```python
from cytearc import PipelineExecutionError

try:
    run = ds.pipeline.run(label="baseline")
except PipelineExecutionError as error:
    failed_run = ds.pipeline.open(run_id=error.run_id)
    report = failed_run.report(format="dict")
```

## Automatic cluster selection

The `cluster_selection` stage scores enabled Leiden resolutions in the coordinates used to build
the graph: PCA or Harmony coordinates, or with `pca_dims=0` the normalized values. Paris remains
`run["paris"]` for diagnosis and comparison; it is never the automatic `run["clusters"]` winner.
Selection uses one deterministic shared sample of at most 10,000 selected cells with seed `4466`.
The sample reserves up to two seeded cells per cluster across every Leiden candidate, then fills
remaining capacity without replacement. The sampled cells' coordinates are read in blocks under the
datastore memory budget, and pairwise distances are computed in chunks of a quarter of the budget,
at most 1 GiB. Before it reads a coordinate, the stage fails when the sampled coordinates, in their
stored dtype and as float64, with one block of the read or with the distance chunks, exceed the
budget: the run raises a `PipelineExecutionError` whose `__cause__` is the `MemoryError`. Reading
a sample spread over a large normalized matrix, as with `pca_dims=0`, decodes most of its stored
chunks.

This silhouette comparison is a reproducible provisional baseline. It is not biological validation
or ground truth. Keep alternative Leiden refs and Paris when the study question needs other
evidence.

The resulting `cluster_selection` artifact records candidate keys and refs, silhouette scores,
the sample definition (`sampleStrategy="sharedClusterQuota"` and `minClusterQuota=2`),
invalid-candidate reasons, deterministic tie order, and the selected key.
`run["clusters"]` is the selected Leiden candidate's exact ref. It is not a copied artifact or metadata
column. If no Leiden candidate can be scored, the stage fails. A single valid Leiden candidate still
receives a decision artifact.

Persisted cluster selection is intentionally pipeline-only. The fixed recipe owns the candidate
set, deterministic sampling policy, and run-ledger decision. Granular callers can evaluate their
own partitions with public APIs such as `evaluate_cluster_separability` in `cytearc.metrics`. There
is no public "best clustering" function and no separate `DataStore` persistence method that could
imply the pipeline policy outside a run. The agent orchestrator is a separate multi-metric
workflow and does not replace this baseline.

## Durable outputs and frozen views

A completed default run exposes these keys in order:

```text
input_cell_selection, analysis_cell_selection, feature_universe, cell_cycle,
highly_variable_features, normalized, pca, ann_index, neighbors,
connectivity_map, embedding_initialization, umap,
leiden_0.5, leiden_0.75, leiden_1.0, leiden_1.25,
paris, cluster_selection, clusters, doublets, markers
```

`harmony` appears after `pca` when enabled. Disabled stages omit their outputs. Normal mapping
operations such as `run["pca"]`, `list(run)`, and `run.items()` are available only on a successfully
completed run.

`run.cells` and `run.features` are narrow read-only views over captured selections, metadata
snapshots, and result artifacts:

```python
view.columns
view.fetch(column)                 # rows selected by the run's stored I
view.fetch_all(column)             # values aligned to the complete stored axis
view.to_pandas_dataframe(columns)  # selected rows
view.head(n=5)
```

Cell `I` is the analysis selection. Feature `I` is backed by the immutable
`run["feature_universe"]` selection created during the input stage, not the live feature `I`
column. The views retain captured names and requested metadata even if live values or live `I`
later change.
They fail closed if ordered cell or feature identities change. There is no live overlay or mutation
surface.

Pipeline execution writes immutable artifacts and run records only. It never inserts result
columns or rewrites live `I`. Plotting, marker loading, and export stay on `DataStore`:

```python
ds.plots.embedding(run=run, layout="umap", color_by="clusters")
markers = ds.markers.load(marker=run["markers"], group_id=0)
adata = ds.to_anndata(run=run)
# adata.obsm["X_umap"] holds frozen UMAP; cluster labels stay in adata.obs
```

The marker table of a run saved before CyteArc 1.0.0 cannot be read, because its `fold_change`
column held sentinels: `markers.load` raises and names
`ds.markers.search(run["clusters"], features=run["feature_universe"])`, which recomputes the
table for the run's clusters. A saved agent result keeps the table that it read, so its analysis
must be run again.

## Open, list, and report runs

A successful optional label is an immutable name for one run:

```python
run = ds.pipeline.open(label="baseline")
same_run = ds.pipeline.open(run_id=run.run_id)

recent = ds.pipeline.list_runs(limit=20)
failed = ds.pipeline.list_runs(status="failed")
interrupted = ds.pipeline.list_runs(status="interrupted")
```

`open` requires exactly one of `run_id` or `label`. Only a successfully completed run acquires its
requested label. Failed and interrupted attempts do not reserve it. `list_runs` returns newest
first and includes all statuses unless filtered. Concurrent finalizers use an atomic label claim,
so at most one completed run can acquire a name. A storage backend without atomic conditional
creation rejects `label=` before a run record or computation starts; unlabeled runs are unaffected.
An unclean incomplete finalizer blocks reuse of its requested label and fails closed.

After confirming that the owner process has stopped, explicitly abandon that exact claim before
retrying the label:

```python
interrupted = ds.pipeline.abandon_label_claim(
    label="baseline",
    run_id=stopped_run_id,
    reason="worker terminated after finalization began",
)
replacement = ds.pipeline.run(label="baseline")
```

The recovery call succeeds only for the exact current owner while `complete=False`, including a
torn terminal payload, and records an `abandoned_label_claim` interruption. It refuses every
`complete=True` owner and has no timeout or automatic stale-process heuristic.

Every status exposes identity, status, and `run.report(format="dict" | "markdown")`. Only a
completed run exposes mapping outputs and frozen views. Reports include stage timing, sampled
process-tree RSS, artifact plans with `created` or `reused` dispositions, failures, interruption
details, and signal-guard availability. RSS peaks are sampled lower bounds. An unavailable
measurement is reported as null with a reason. RSS comes from the Linux `/proc` filesystem; on
platforms without it, such as macOS and Windows, nothing is sampled, so `sampleCount` is 0.

A hard process death can leave `complete=False`; this is reported as an unclean incomplete run.
There is no resume, repair, or same-ID retry. A new invocation may reuse only complete artifacts.

### Runs saved by other releases

A run completed under any 1.x release reopens under every later 1.x release. `open` by label or
run ID, `list_runs`, `run[key]`, `run.report()`, and the frozen `run.cells` and `run.features`
views work as long as the artifacts the run references exist and satisfy their payload contracts.
Running a saved configuration again recomputes a stage whose operation gained a revision that
applies to it, or whose recorded provenance a later release changed; it never modifies the earlier
run. Lineage reports of the earlier run's outputs mark those that a revision supersedes `stale`.

A run that a newer CyteArc saved with record fields this release does not know fails closed: opening
it by run ID raises a `ValueError` that names those fields. Upgrade CyteArc to open it. Within 1.x,
releases only add record fields, so runs from earlier releases never meet this error. A 2.0 release
may change run records and lists each change in its compatibility inventory.

Run and stage records are otherwise strict: a missing or unknown field, or a malformed value,
fails closed. `list_runs` and open-by-label skip records they cannot read so healthy runs remain
accessible, and log a warning that names each skipped run ID and why it could not be read. Opening
that run by its exact ID raises its error.

## Graceful interruption and callbacks

On the main interpreter thread, a pipeline temporarily installs cooperative handlers for available
`SIGTERM`, `SIGINT`, and `SIGHUP`, while respecting signals already set to ignore. The first signal
requests shutdown at a safe checkpoint. The pipeline persists completed or interrupted stage and
run state, invokes interruption callbacks, restores prior handlers, and propagates the original
signal behavior. A second catchable termination signal escalates immediately. Non-main-thread runs
record that signal protection was unavailable.

`KeyboardInterrupt` and an escaped `asyncio.CancelledError` use the same durable interruption
boundary. Ordinary stage exceptions produce a failed run instead.
Run and stage records hold at most 512 characters of an error or interruption message. A longer
message keeps its first 509 characters followed by `...`; the raised exception keeps its full
message.
If a termination signal races with an ordinary failure or the final successful handoff, the
pipeline preserves that durable outcome and still propagates the pending signal after cleanup.

`SIGKILL`, `SIGSTOP`, out-of-memory termination, power loss, and expired shutdown grace periods
cannot perform cleanup. Their durable contract is an incomplete artifact or run, followed by a new
invocation that reuses only complete artifacts.

Callbacks provide serialized progress notifications without controlling execution:

```python
from cytearc.datastore.pipeline_accessor import PipelineEvent

events: list[tuple[str, str]] = []


def record_event(event: PipelineEvent) -> None:
    events.append((event.kind, event.stage))


run = ds.pipeline.run(callback=record_event)
```

Enabled stages emit `stage_started` once their record is written, then `stage_completed`,
`stage_failed`, or `stage_interrupted` once their outcome is recorded, before the next stage starts.
A handled interruption also emits `pipeline_interrupted` after durable state is written. A skipped
stage is present in the report and emits no stage event, unless an error or interruption arrives
between writing its record and recording it as skipped; it is then recorded as failed or interrupted
and emits `stage_failed` or `stage_interrupted` without a preceding `stage_started`. Callback errors
are logged and cannot block durable status.

## Public types

```{eval-rst}
.. autoclass:: cytearc.datastore.pipeline_accessor.PipelineAccessor
    :members: run, open, list_runs, abandon_label_claim

.. autoclass:: cytearc.datastore.pipeline_accessor.PipelineEvent
    :members:

.. autoclass:: cytearc.PipelineRun
    :members:

.. autoclass:: cytearc.PipelineExecutionError
    :members:
```

See [Quick start](../quickstart) for an executable first run and {doc}`graph_construction` for
explicit stage-by-stage APIs.