API reference

adag

adag: a cached, content-addressed DAG engine for Python data pipelines.

This module exports every public name of the package, including the lower-level pieces (executors, dispatchers, ResolveContext, walk).

JsonValue

JsonValue = (
    bool
    | int
    | float
    | str
    | Sequence[JsonValue]
    | Mapping[str, JsonValue]
    | None
)

A value json.dumps accepts: the type of Pipeline.metadata.

Sequence and Mapping (not list and dict) so a dict[str, int] is accepted under strict type checking. Narrow it (isinstance) before you use a part of it.

CachedStep dataclass

Bases: Step[CT]

Deprecated: a Step with direct save/load/is_cached keyed on the step hash.

Source code in src/adag/base.py

class CachedStep(Step[CT]):
    """Deprecated: a ``Step`` with direct ``save``/``load``/``is_cached`` keyed on the step hash."""

    def __init_subclass__(cls, **kwargs: Any) -> None:
        super().__init_subclass__(**kwargs)
        if any(get_origin(base) is CachedStep for base in getattr(cls, "__orig_bases__", [])):
            warnings.warn(
                f"{cls.__name__}: CachedStep is deprecated and is removed in adag 1.0. Subclass "
                "Step[...] instead; for a custom cache format override save_result/load_result "
                "(docs/caching.md).",
                AdagDeprecationWarning,
                stacklevel=2,
            )

    def output_path(self, cache_dir: Path) -> Path | None:
        """Return ``cache_dir/<step hash>``, or ``None`` when an input is missing."""
        h = self.content_hash()
        if h is None:
            return None
        return cache_dir.joinpath(h)

    def is_cached(self, cache_dir: Path) -> bool:
        """Return whether the output and its ``.done`` marker exist and ``validate`` passes."""
        output = self.output_path(cache_dir)
        if output is None or not output.exists():
            return False
        if not output.parent.joinpath(output.name + ".done").exists():
            return False
        return self.validate(output)

    def is_partial(self, cache_dir: Path) -> bool:
        """Return whether the output exists without its ``.done`` marker (a crash mid-save)."""
        output = self.output_path(cache_dir)
        if output is None:
            return False
        return output.exists() and not output.parent.joinpath(output.name + ".done").exists()

    def save(self, result: CT, cache_dir: Path) -> None:
        """Write ``result`` as JSON to ``output_path``, then the ``.done`` marker. Both atomic.

        Raises:
            ValueError: an input is missing, so the step has no hash.
        """
        output = self.output_path(cache_dir)
        if output is None:
            raise ValueError(f"Step '{type(self).__name__}' is BLOCKED: inputs missing.")
        atomic_write_text(output, result.model_dump_json(indent=2))
        atomic_write_text(output.parent.joinpath(output.name + ".done"), "")

    def load(self, cache_dir: Path) -> CT:
        """Read the result from ``output_path`` as JSON.

        Raises:
            ValueError: no output, or the step has no ``BaseModel`` result type.
        """
        output = self.output_path(cache_dir)
        result_type = self.result_model()
        if output is None or not output.exists():
            raise ValueError(f"Output for '{type(self).__name__}' not found.")
        if result_type is None:
            raise ValueError(f"Step '{type(self).__name__}' has no result_type.")
        return result_type.model_validate_json(output.read_text())  # pyright: ignore[reportReturnType]

    def validate(self, path: Path) -> bool:
        """Return whether ``path`` parses as the result type. An empty file is not valid."""
        result_type = self.result_model()
        if result_type is None:
            return False
        try:
            content = path.read_text()
            if not content.strip():
                return False
            result_type.model_validate_json(content)
        except Exception:  # any read or parse error means "not valid"
            return False
        return True

    def metadata(self, result: CT) -> dict[str, Any]:
        """Return extra manifest fields for ``result``. Override to add some; default is none."""
        del result
        return {}

output_path

output_path(cache_dir: Path) -> Path | None

Return cache_dir/<step hash>, or None when an input is missing.

Source code in src/adag/base.py

def output_path(self, cache_dir: Path) -> Path | None:
    """Return ``cache_dir/<step hash>``, or ``None`` when an input is missing."""
    h = self.content_hash()
    if h is None:
        return None
    return cache_dir.joinpath(h)

is_cached

is_cached(cache_dir: Path) -> bool

Return whether the output and its .done marker exist and validate passes.

Source code in src/adag/base.py

def is_cached(self, cache_dir: Path) -> bool:
    """Return whether the output and its ``.done`` marker exist and ``validate`` passes."""
    output = self.output_path(cache_dir)
    if output is None or not output.exists():
        return False
    if not output.parent.joinpath(output.name + ".done").exists():
        return False
    return self.validate(output)

is_partial

is_partial(cache_dir: Path) -> bool

Return whether the output exists without its .done marker (a crash mid-save).

Source code in src/adag/base.py

def is_partial(self, cache_dir: Path) -> bool:
    """Return whether the output exists without its ``.done`` marker (a crash mid-save)."""
    output = self.output_path(cache_dir)
    if output is None:
        return False
    return output.exists() and not output.parent.joinpath(output.name + ".done").exists()

save

save(result: CT, cache_dir: Path) -> None

Write result as JSON to output_path, then the .done marker. Both atomic.

RAISES	DESCRIPTION
ValueError	an input is missing, so the step has no hash.

Source code in src/adag/base.py

def save(self, result: CT, cache_dir: Path) -> None:
    """Write ``result`` as JSON to ``output_path``, then the ``.done`` marker. Both atomic.

    Raises:
        ValueError: an input is missing, so the step has no hash.
    """
    output = self.output_path(cache_dir)
    if output is None:
        raise ValueError(f"Step '{type(self).__name__}' is BLOCKED: inputs missing.")
    atomic_write_text(output, result.model_dump_json(indent=2))
    atomic_write_text(output.parent.joinpath(output.name + ".done"), "")

load

load(cache_dir: Path) -> CT

Read the result from output_path as JSON.

RAISES	DESCRIPTION
ValueError	no output, or the step has no BaseModel result type.

Source code in src/adag/base.py

def load(self, cache_dir: Path) -> CT:
    """Read the result from ``output_path`` as JSON.

    Raises:
        ValueError: no output, or the step has no ``BaseModel`` result type.
    """
    output = self.output_path(cache_dir)
    result_type = self.result_model()
    if output is None or not output.exists():
        raise ValueError(f"Output for '{type(self).__name__}' not found.")
    if result_type is None:
        raise ValueError(f"Step '{type(self).__name__}' has no result_type.")
    return result_type.model_validate_json(output.read_text())  # pyright: ignore[reportReturnType]

validate

validate(path: Path) -> bool

Return whether path parses as the result type. An empty file is not valid.

Source code in src/adag/base.py

def validate(self, path: Path) -> bool:
    """Return whether ``path`` parses as the result type. An empty file is not valid."""
    result_type = self.result_model()
    if result_type is None:
        return False
    try:
        content = path.read_text()
        if not content.strip():
            return False
        result_type.model_validate_json(content)
    except Exception:  # any read or parse error means "not valid"
        return False
    return True

metadata

metadata(result: CT) -> dict[str, Any]

Return extra manifest fields for result. Override to add some; default is none.

Source code in src/adag/base.py

def metadata(self, result: CT) -> dict[str, Any]:
    """Return extra manifest fields for ``result``. Override to add some; default is none."""
    del result
    return {}

BatchItem dataclass

A single work unit in a BatchStep.

key: cache key for this item (determines cache identity) data: arbitrary payload passed to process_item()

Source code in src/adag/batch.py

@dataclass(frozen=True)
class BatchItem:
    """A single work unit in a BatchStep.

    key:  cache key for this item (determines cache identity)
    data: arbitrary payload passed to process_item()
    """

    key: str
    data: Any = None

BatchStep dataclass

Bases: Step[T], Generic[T, R]

A step that processes items individually with per-item disk caching.

Subclasses implement three methods:

1. items(**inputs): declare work units and their cache keys
2. process_item(item, **inputs): handle one item
3. assemble(results, **inputs): combine all results into final output

The framework automatically: - Partitions items into cached/uncached - Skips already-processed items - Saves each result to disk immediately after processing - Logs progress (cached/total, processing each item) - Only marks .done when ALL items are complete

On crash: restart picks up where it left off, only uncached items run.

Per-item cache key derivation. The framework auto-derives each item's cache key from sha256(step_signature + inputs_signature + item_data). The inputs signature includes every kwarg passed to :meth:execute except those listed in :attr:items_source_keys. Those are treated as containers of items rather than shared parameters. Subclasses that take a large input list and extract items from it should set items_source_keys to the kwarg name(s) so that adding or removing items doesn't invalidate every other item's cache::

class Greet(BatchStep[Greetings, Greeting]):
    items_source_keys = ("names",)  # the list of items
    # An input that is not listed here (a prompt, for example) is shared by
    # every item, so a change to it invalidates all per-item caches.

Source code in src/adag/batch.py

class BatchStep(Step[T], Generic[T, R]):
    """A step that processes items individually with per-item disk caching.

    Subclasses implement three methods:

    1. ``items(**inputs)``: declare work units and their cache keys
    2. ``process_item(item, **inputs)``: handle one item
    3. ``assemble(results, **inputs)``: combine all results into final output

    The framework automatically:
    - Partitions items into cached/uncached
    - Skips already-processed items
    - Saves each result to disk immediately after processing
    - Logs progress (cached/total, processing each item)
    - Only marks .done when ALL items are complete

    On crash: restart picks up where it left off, only uncached items run.

    **Per-item cache key derivation.** The framework auto-derives each
    item's cache key from ``sha256(step_signature + inputs_signature +
    item_data)``. The inputs signature includes every kwarg passed to
    :meth:`execute` except those listed in :attr:`items_source_keys`.
    Those are treated as containers of items rather than shared
    parameters. Subclasses that take a large input list and extract items
    from it should set ``items_source_keys`` to the kwarg name(s) so that
    adding or removing items doesn't invalidate every other item's cache::

        class Greet(BatchStep[Greetings, Greeting]):
            items_source_keys = ("names",)  # the list of items
            # An input that is not listed here (a prompt, for example) is shared by
            # every item, so a change to it invalidates all per-item caches.
    """

    item_type: type[BaseModel]  # auto-resolved from BatchStep[T, R]

    # Names of input kwargs that are CONTAINERS of items rather than
    # shared parameters. Excluded from per-item cache key derivation.
    # Default empty: ALL inputs participate in the key (safe but
    # potentially over-invalidating). Subclasses with a large source-list
    # should override.
    items_source_keys: ClassVar[tuple[str, ...]] = ()

    def __init_subclass__(cls, **kwargs: Any) -> None:
        super().__init_subclass__(**kwargs)
        for parent in getattr(cls, "__orig_bases__", []):
            origin = get_origin(parent)
            if origin is BatchStep:
                args = get_args(parent)
                if len(args) >= 2:  # noqa: PLR2004  (BatchStep[T, Item]: the item type is the 2nd arg)
                    cls.item_type = args[1]
                    return

    def items_data(self, **inputs: Any) -> list[Any]:
        """Declare the per-item DATA only, framework derives the cache key.

        Recommended override path. The default ``items()`` (below) calls
        this and wraps each entry in a ``BatchItem`` with an auto-computed
        key that includes the step's content_hash + the item's repr.

        Returns the list of data objects (any type) that
        ``process_item(item, ...)`` will receive as ``item.data``.

        Override this method when you don't need custom key topology.
        Then no cache-poisoning bug from manual key composition is possible.
        """
        raise NotImplementedError(
            f"{type(self).__name__} must override either `items_data()` (framework derives keys) or `items()` (full control)."
        )

    def items(self, **inputs: Any) -> list[BatchItem]:
        """Declare the work units. Called once at the start.

        Default implementation calls ``items_data(**inputs)`` and assigns
        each item an auto-generated cache key derived from
        ``(step.content_hash, **inputs, item_data)``. The inputs signature
        is included so that step kwargs which AFFECT the per-item LLM call
        (e.g. a notes payload passed alongside the report shortlists)
        invalidate item caches when they change, even though they are
        not step dataclass fields.

        Override this only when you need custom key topology (e.g.
        deliberately sharing a cached result across two slightly-different
        inputs).
        """
        step_sig = self._step_signature()
        inputs_sig = self._inputs_signature(inputs)
        return [
            BatchItem(key=self._auto_item_key(step_sig, datum, inputs_sig), data=datum)
            for datum in self.items_data(**inputs)
        ]

    def _step_signature(self) -> str:
        """A short hex digest of the step's identity for use in item keys.

        Falls back to the class kind only if ``content_hash()`` is
        blocked by an unsatisfied upstream (e.g. a missing FileSource).
        Whenever the step's output semantics change, ``content_hash()``
        changes, and so does every item's auto-generated cache key.
        """
        return self.content_hash() or type(self).kind or type(self).__name__

    def _inputs_signature(self, inputs: dict[str, Any]) -> str:
        """Hash all BaseModel/list/scalar inputs into a single digest.

        The point of this hash is to catch upstream-result changes (like a
        notes payload, a scoped item list, a config blob) that affect
        what ``process_item`` produces but are not step dataclass fields.
        Without this, a BatchStep's per-item cache would silently return
        stale results whenever an input kwarg changed but step config and
        per-item data did not.

        Inputs named in :attr:`items_source_keys` are skipped. They are
        containers from which items are extracted, not shared parameters.
        Including them would invalidate every item's cache when the list
        of items changed (e.g. one new report added).
        """
        skip = set(type(self).items_source_keys)
        h = hashlib.sha256()
        for name in sorted(inputs.keys()):
            if name in skip:
                continue
            val = inputs[name]
            h.update(name.encode())
            h.update(b"=")
            if isinstance(val, BaseModel):
                h.update(val.model_dump_json().encode())
            elif isinstance(val, (str, int, float, bool)) or val is None:
                h.update(repr(val).encode())
            elif isinstance(val, (list, tuple)):
                try:
                    h.update(json.dumps(val, sort_keys=True, default=str).encode())
                except (TypeError, ValueError):
                    # repr() handles any object; pyright over-narrows the
                    # generic params here because dict was dict[str, Any].
                    h.update(repr(val).encode())  # pyright: ignore[reportUnknownArgumentType]
            else:
                # Unknown type: fall back to repr for stability.
                h.update(repr(val).encode())
            h.update(b"|")
        return h.hexdigest()[:16]

    def _auto_item_key(self, step_sig: str, item_data: Any, inputs_sig: str = "") -> str:
        """Derive a stable cache key from ``(step_sig, inputs_sig, item_data)``.

        ``item_data`` can be any object. We serialize it deterministically
        (Pydantic JSON if possible, then a stable repr fallback). The
        digest is truncated to 16 hex chars to match ItemCache.key().

        ``inputs_sig`` is the digest of all ``**inputs`` kwargs from
        :meth:`execute`. Included so that changing any input invalidates
        per-item caches: see :meth:`_inputs_signature`.
        """
        h = hashlib.sha256()
        h.update(step_sig.encode())
        h.update(b"|")
        h.update(inputs_sig.encode())
        h.update(b"|")
        if isinstance(item_data, BaseModel):
            h.update(item_data.model_dump_json().encode())
        else:
            try:
                h.update(json.dumps(item_data, sort_keys=True, default=str).encode())
            except (TypeError, ValueError):
                h.update(repr(item_data).encode())
        return h.hexdigest()[:16]

    @abstractmethod
    def process_item(self, item: BatchItem, **inputs: Any) -> R:
        """Process a single item. Default implementation called by process_batch().

        Override this for simple one-at-a-time processing.
        Receives the BatchItem plus all loaded input kwargs.
        """
        ...

    def process_batch(self, items: list[BatchItem], cache: ItemCache[R], **inputs: Any) -> None:
        """Process a batch of uncached items, saving each result to cache.

        Override this to control batching strategy (concurrent LLM calls,
        chunked API requests, etc.). The default processes one at a time.

        Each result MUST be saved to cache immediately via ``cache.save()``
        so a crash loses at most one in-flight item.
        """
        for i, item in enumerate(items, start=1):
            emit(
                "batch.processing",
                step=type(self).__name__,
                item=item.key[:12],
                progress=f"{i}/{len(items)}",
            )
            # WAL: write the `.inflight` sentinel before the call starts.
            # If `process_item` raises or the process is killed, the
            # marker survives so the next resume can surface "you lost
            # work on these items". `cache.save()` clears it on success.
            cache.mark_inflight(item.key)
            result = self.process_item(item, **inputs)
            cache.save(item.key, result)

    @abstractmethod
    def assemble(self, results: list[R], **inputs: Any) -> T:
        """Combine all per-item results into the final step result.

        Called after all items (cached + freshly processed) are available.
        Receives the ordered list of results plus all loaded input kwargs.
        """
        ...

    def item_cache_namespace(self) -> str:
        """Namespace for the ItemCache. Default: class name in snake_case."""
        name = type(self).__name__
        result: list[str] = []
        for i, c in enumerate(name):
            if c.isupper() and i > 0:
                result.append("_")
            result.append(c.lower())
        return "".join(result)

    def run(self, **inputs: Any) -> T:  # pyright: ignore[reportUnusedParameter]
        """Raise: a BatchStep is driven through ``execute()``, never ``run()``."""
        raise RuntimeError("BatchStep uses execute(context, ...); do not call run() directly.")

    def execute(self, context: ResolveContext, /, **inputs: Any) -> T:
        """Orchestrate: items → partition → process uncached → assemble.

        Called by StepResult.resolve with the current ResolveContext so the
        per-item cache can live under the pipeline's cache directory.
        Subclasses implement items/process_item/assemble, not this.
        """
        if context.cache_dir is None and context.store is None:
            raise RuntimeError(
                f"{type(self).__name__} requires a Pipeline cache_dir or store for per-item caching."
            )
        cache_dir = context.cache_dir or Path()  # unused when the store is set

        namespace = self.item_cache_namespace()
        item_schema: Any = self.item_type
        cache: ItemCache[R] = ItemCache(cache_dir, namespace, item_schema, store=context.store)

        # Step 1: Declare items
        all_items = self.items(**inputs)
        keys = [item.key for item in all_items]

        emit(
            "batch.start",
            step=type(self).__name__,
            total=len(all_items),
            namespace=namespace,
        )

        # Step 2: Partition cached/uncached
        _cached, uncached_indices = cache.partition(keys)
        n_cached = len(all_items) - len(uncached_indices)

        if n_cached > 0:
            emit(
                "batch.resuming",
                step=type(self).__name__,
                cached=n_cached,
                remaining=len(uncached_indices),
            )

        # Step 3: Process uncached items via process_batch()
        # The step decides internally how to batch (concurrent, chunked, etc.)
        uncached_items = [all_items[idx] for idx in uncached_indices]
        if uncached_items:
            self.process_batch(uncached_items, cache, **inputs)

        # Step 4: Collect all results in order.
        # Tolerate per-item failures: LLM batches in particular routinely have
        # a small percentage fail due to rate limits / transient errors, and
        # the successful majority is worth keeping. Subclasses that need
        # strict behavior should validate the assembled result themselves.
        all_results = cache.collect(keys, allow_missing=True)

        emit(
            "batch.done",
            step=type(self).__name__,
            total=len(all_results),
            processed=len(uncached_indices),
            from_cache=n_cached,
        )

        # Step 5: Assemble final result
        return self.assemble(all_results, **inputs)

    def run_concurrent(self, *, cache_dir: Any, **inputs: Any) -> T:
        """Drive a BatchStep outside a Pipeline.resolve() context.

        Use this when another Step's ``run()`` body needs to invoke a
        BatchStep over derived data (e.g. running Stage-2 classification
        from inside a larger Step). This helper builds a minimal
        ResolveContext rooted at ``cache_dir`` and dispatches through
        :meth:`execute`, which guarantees:

        * Item cache keys are auto-derived from
          ``sha256(step.content_hash() + item.model_dump_json())[:16]``,
          so any change to step configuration (model, max_candidates,
          notes, etc.) OR to a single item's input invalidates that
          item's cache entry. No stale results.
        * Partition / process_batch / collect / assemble run in order.

        DO NOT build ``BatchItem(key=..., data=...)`` manually and call
        ``process_batch(items, cache, **inputs)`` directly. That bypasses
        the framework's content-aware key derivation. Custom keys that
        encode only part of the inputs (e.g. external_id but not the
        actual data being sent to a model) silently return stale results
        when the un-encoded inputs change. ``run_concurrent`` exists
        specifically to make this footgun harder to find.
        """
        from adag.deferred import ResolveContext

        context = ResolveContext(cache_dir=cache_dir, params={}, on_progress=None)
        return self.execute(context, **inputs)

items_data

items_data(**inputs: Any) -> list[Any]

Declare the per-item DATA only, framework derives the cache key.

Recommended override path. The default items() (below) calls this and wraps each entry in a BatchItem with an auto-computed key that includes the step's content_hash + the item's repr.

Returns the list of data objects (any type) that process_item(item, ...) will receive as item.data.

Override this method when you don't need custom key topology. Then no cache-poisoning bug from manual key composition is possible.

Source code in src/adag/batch.py

def items_data(self, **inputs: Any) -> list[Any]:
    """Declare the per-item DATA only, framework derives the cache key.

    Recommended override path. The default ``items()`` (below) calls
    this and wraps each entry in a ``BatchItem`` with an auto-computed
    key that includes the step's content_hash + the item's repr.

    Returns the list of data objects (any type) that
    ``process_item(item, ...)`` will receive as ``item.data``.

    Override this method when you don't need custom key topology.
    Then no cache-poisoning bug from manual key composition is possible.
    """
    raise NotImplementedError(
        f"{type(self).__name__} must override either `items_data()` (framework derives keys) or `items()` (full control)."
    )

items

items(**inputs: Any) -> list[BatchItem]

Declare the work units. Called once at the start.

Default implementation calls items_data(**inputs) and assigns each item an auto-generated cache key derived from (step.content_hash, **inputs, item_data). The inputs signature is included so that step kwargs which AFFECT the per-item LLM call (e.g. a notes payload passed alongside the report shortlists) invalidate item caches when they change, even though they are not step dataclass fields.

Override this only when you need custom key topology (e.g. deliberately sharing a cached result across two slightly-different inputs).

Source code in src/adag/batch.py

def items(self, **inputs: Any) -> list[BatchItem]:
    """Declare the work units. Called once at the start.

    Default implementation calls ``items_data(**inputs)`` and assigns
    each item an auto-generated cache key derived from
    ``(step.content_hash, **inputs, item_data)``. The inputs signature
    is included so that step kwargs which AFFECT the per-item LLM call
    (e.g. a notes payload passed alongside the report shortlists)
    invalidate item caches when they change, even though they are
    not step dataclass fields.

    Override this only when you need custom key topology (e.g.
    deliberately sharing a cached result across two slightly-different
    inputs).
    """
    step_sig = self._step_signature()
    inputs_sig = self._inputs_signature(inputs)
    return [
        BatchItem(key=self._auto_item_key(step_sig, datum, inputs_sig), data=datum)
        for datum in self.items_data(**inputs)
    ]

process_item abstractmethod

process_item(item: BatchItem, **inputs: Any) -> R

Process a single item. Default implementation called by process_batch().

Override this for simple one-at-a-time processing. Receives the BatchItem plus all loaded input kwargs.

Source code in src/adag/batch.py

@abstractmethod
def process_item(self, item: BatchItem, **inputs: Any) -> R:
    """Process a single item. Default implementation called by process_batch().

    Override this for simple one-at-a-time processing.
    Receives the BatchItem plus all loaded input kwargs.
    """
    ...

process_batch

process_batch(
    items: list[BatchItem],
    cache: ItemCache[R],
    **inputs: Any,
) -> None

Process a batch of uncached items, saving each result to cache.

Override this to control batching strategy (concurrent LLM calls, chunked API requests, etc.). The default processes one at a time.

Each result MUST be saved to cache immediately via cache.save() so a crash loses at most one in-flight item.

Source code in src/adag/batch.py

def process_batch(self, items: list[BatchItem], cache: ItemCache[R], **inputs: Any) -> None:
    """Process a batch of uncached items, saving each result to cache.

    Override this to control batching strategy (concurrent LLM calls,
    chunked API requests, etc.). The default processes one at a time.

    Each result MUST be saved to cache immediately via ``cache.save()``
    so a crash loses at most one in-flight item.
    """
    for i, item in enumerate(items, start=1):
        emit(
            "batch.processing",
            step=type(self).__name__,
            item=item.key[:12],
            progress=f"{i}/{len(items)}",
        )
        # WAL: write the `.inflight` sentinel before the call starts.
        # If `process_item` raises or the process is killed, the
        # marker survives so the next resume can surface "you lost
        # work on these items". `cache.save()` clears it on success.
        cache.mark_inflight(item.key)
        result = self.process_item(item, **inputs)
        cache.save(item.key, result)

assemble abstractmethod

assemble(results: list[R], **inputs: Any) -> T

Combine all per-item results into the final step result.

Called after all items (cached + freshly processed) are available. Receives the ordered list of results plus all loaded input kwargs.

Source code in src/adag/batch.py

@abstractmethod
def assemble(self, results: list[R], **inputs: Any) -> T:
    """Combine all per-item results into the final step result.

    Called after all items (cached + freshly processed) are available.
    Receives the ordered list of results plus all loaded input kwargs.
    """
    ...

item_cache_namespace

item_cache_namespace() -> str

Namespace for the ItemCache. Default: class name in snake_case.

Source code in src/adag/batch.py

def item_cache_namespace(self) -> str:
    """Namespace for the ItemCache. Default: class name in snake_case."""
    name = type(self).__name__
    result: list[str] = []
    for i, c in enumerate(name):
        if c.isupper() and i > 0:
            result.append("_")
        result.append(c.lower())
    return "".join(result)

run

run(**inputs: Any) -> T

Raise: a BatchStep is driven through execute(), never run().

Source code in src/adag/batch.py

def run(self, **inputs: Any) -> T:  # pyright: ignore[reportUnusedParameter]
    """Raise: a BatchStep is driven through ``execute()``, never ``run()``."""
    raise RuntimeError("BatchStep uses execute(context, ...); do not call run() directly.")

execute

execute(context: ResolveContext, /, **inputs: Any) -> T

Orchestrate: items → partition → process uncached → assemble.

Called by StepResult.resolve with the current ResolveContext so the per-item cache can live under the pipeline's cache directory. Subclasses implement items/process_item/assemble, not this.

Source code in src/adag/batch.py

def execute(self, context: ResolveContext, /, **inputs: Any) -> T:
    """Orchestrate: items → partition → process uncached → assemble.

    Called by StepResult.resolve with the current ResolveContext so the
    per-item cache can live under the pipeline's cache directory.
    Subclasses implement items/process_item/assemble, not this.
    """
    if context.cache_dir is None and context.store is None:
        raise RuntimeError(
            f"{type(self).__name__} requires a Pipeline cache_dir or store for per-item caching."
        )
    cache_dir = context.cache_dir or Path()  # unused when the store is set

    namespace = self.item_cache_namespace()
    item_schema: Any = self.item_type
    cache: ItemCache[R] = ItemCache(cache_dir, namespace, item_schema, store=context.store)

    # Step 1: Declare items
    all_items = self.items(**inputs)
    keys = [item.key for item in all_items]

    emit(
        "batch.start",
        step=type(self).__name__,
        total=len(all_items),
        namespace=namespace,
    )

    # Step 2: Partition cached/uncached
    _cached, uncached_indices = cache.partition(keys)
    n_cached = len(all_items) - len(uncached_indices)

    if n_cached > 0:
        emit(
            "batch.resuming",
            step=type(self).__name__,
            cached=n_cached,
            remaining=len(uncached_indices),
        )

    # Step 3: Process uncached items via process_batch()
    # The step decides internally how to batch (concurrent, chunked, etc.)
    uncached_items = [all_items[idx] for idx in uncached_indices]
    if uncached_items:
        self.process_batch(uncached_items, cache, **inputs)

    # Step 4: Collect all results in order.
    # Tolerate per-item failures: LLM batches in particular routinely have
    # a small percentage fail due to rate limits / transient errors, and
    # the successful majority is worth keeping. Subclasses that need
    # strict behavior should validate the assembled result themselves.
    all_results = cache.collect(keys, allow_missing=True)

    emit(
        "batch.done",
        step=type(self).__name__,
        total=len(all_results),
        processed=len(uncached_indices),
        from_cache=n_cached,
    )

    # Step 5: Assemble final result
    return self.assemble(all_results, **inputs)

run_concurrent

run_concurrent(*, cache_dir: Any, **inputs: Any) -> T

Drive a BatchStep outside a Pipeline.resolve() context.

Use this when another Step's run() body needs to invoke a BatchStep over derived data (e.g. running Stage-2 classification from inside a larger Step). This helper builds a minimal ResolveContext rooted at cache_dir and dispatches through :meth:execute, which guarantees:

• Item cache keys are auto-derived from sha256(step.content_hash() + item.model_dump_json())[:16], so any change to step configuration (model, max_candidates, notes, etc.) OR to a single item's input invalidates that item's cache entry. No stale results.
• Partition / process_batch / collect / assemble run in order.

DO NOT build BatchItem(key=..., data=...) manually and call process_batch(items, cache, **inputs) directly. That bypasses the framework's content-aware key derivation. Custom keys that encode only part of the inputs (e.g. external_id but not the actual data being sent to a model) silently return stale results when the un-encoded inputs change. run_concurrent exists specifically to make this footgun harder to find.

Source code in src/adag/batch.py

def run_concurrent(self, *, cache_dir: Any, **inputs: Any) -> T:
    """Drive a BatchStep outside a Pipeline.resolve() context.

    Use this when another Step's ``run()`` body needs to invoke a
    BatchStep over derived data (e.g. running Stage-2 classification
    from inside a larger Step). This helper builds a minimal
    ResolveContext rooted at ``cache_dir`` and dispatches through
    :meth:`execute`, which guarantees:

    * Item cache keys are auto-derived from
      ``sha256(step.content_hash() + item.model_dump_json())[:16]``,
      so any change to step configuration (model, max_candidates,
      notes, etc.) OR to a single item's input invalidates that
      item's cache entry. No stale results.
    * Partition / process_batch / collect / assemble run in order.

    DO NOT build ``BatchItem(key=..., data=...)`` manually and call
    ``process_batch(items, cache, **inputs)`` directly. That bypasses
    the framework's content-aware key derivation. Custom keys that
    encode only part of the inputs (e.g. external_id but not the
    actual data being sent to a model) silently return stale results
    when the un-encoded inputs change. ``run_concurrent`` exists
    specifically to make this footgun harder to find.
    """
    from adag.deferred import ResolveContext

    context = ResolveContext(cache_dir=cache_dir, params={}, on_progress=None)
    return self.execute(context, **inputs)

Config

Bases: Generic[T]

Wrap execution-only parameters that do NOT affect the output hash.

Usage in step fields::

@dataclass(frozen=True)
class FindCandidates(Step[Candidates]):
    model: str = "my-model"                      # hashed, affects output
    concurrency: Config[int] = Config(20)       # NOT hashed, execution only
    batch_size: Config[int] = Config(256)       # NOT hashed, execution only
    tracker: Config[Tracker | None] = Config(None)

In run(), the value is unwrapped automatically::

def run(self, model: str, concurrency: int, batch_size: int) -> Candidates:
    ...

Generic over T so you can wrap anything an execution-time knob can hold (primitives, objects, optional handles). The framework's content_hash() skips Config fields entirely, so the wrapped type's hashability doesn't matter.

Source code in src/adag/config.py

class Config(Generic[T]):
    """Wrap execution-only parameters that do NOT affect the output hash.

    Usage in step fields::

        @dataclass(frozen=True)
        class FindCandidates(Step[Candidates]):
            model: str = "my-model"                      # hashed, affects output
            concurrency: Config[int] = Config(20)       # NOT hashed, execution only
            batch_size: Config[int] = Config(256)       # NOT hashed, execution only
            tracker: Config[Tracker | None] = Config(None)

    In ``run()``, the value is unwrapped automatically::

        def run(self, model: str, concurrency: int, batch_size: int) -> Candidates:
            ...

    Generic over ``T`` so you can wrap anything an execution-time knob
    can hold (primitives, objects, optional handles). The framework's
    ``content_hash()`` skips ``Config`` fields entirely, so the wrapped
    type's hashability doesn't matter.
    """

    __slots__ = ("value",)

    # Stored as Any so isinstance(x, Config) narrowing doesn't leave
    # `.value` as Unknown (which then poisons every downstream access).
    # Users get their declared T back through how Step.resolve unwraps
    # Config fields when calling run(): the field's declared
    # ``Config[T]`` gives the run() parameter its T-typed kwarg
    # automatically.
    value: Any

    def __init__(self, value: T) -> None:
        self.value = value

    def __repr__(self) -> str:
        return f"Config({self.value!r})"

    def __eq__(self, other: object) -> bool:
        if not isinstance(other, Config):
            return NotImplemented
        return self.value == other.value

    def __hash__(self) -> int:
        try:
            return hash(self.value)
        except TypeError:
            # Unhashable wrapped value (mutable container, custom class
            # without __hash__). Fall back to identity. Config is mostly
            # used as a frozen-dataclass field where two instances
            # holding equivalent values are constructed once at the call
            # site, so identity-hash is sufficient.
            return id(self.value)

AsyncExecutor

Bases: Protocol

The async counterpart of Executor: decides where a step runs on the ASYNC resolve path (StepResult.aresolve / Pipeline.aexecute). Carried on ResolveContext.async_executor.

Mirrors Executor exactly except run is awaitable, so a step body that awaits I/O (a DB session, an HTTP/LLM call) runs ON the event loop, no thread offload, so the event loop stays responsive and cancellation propagates through the awaits. The default LocalAsyncExecutor runs the step in-process; a remote async executor would dispatch by content hash and await the result.

Source code in src/adag/deferred.py

class AsyncExecutor(Protocol):
    """The async counterpart of ``Executor``: decides *where* a step runs on the ASYNC resolve path
    (``StepResult.aresolve`` / ``Pipeline.aexecute``). Carried on ``ResolveContext.async_executor``.

    Mirrors ``Executor`` exactly except ``run`` is awaitable, so a step body that ``await``s I/O (a DB
    session, an HTTP/LLM call) runs ON the event loop, no thread offload, so the event loop stays
    responsive and cancellation propagates through the awaits. The default ``LocalAsyncExecutor`` runs the
    step in-process; a remote async executor would dispatch by content hash and await the result.
    """

    async def run(
        self, step: Step[Any], context: ResolveContext, inputs: dict[str, Any]
    ) -> Any: ...

BranchDispatcher

Bases: Protocol

Decides where the branches of one fan-out are resolved.

FanOutStep calls dispatch_all once per fan-out with every branch. A dispatcher returns {branch.key: result}, or raises FanOutBranchError naming the failed keys. A worker writes its result to the shared cache under branch.branch_hash, and the dispatcher loads it from there. The default InlineDispatcher resolves in-process.

Source code in src/adag/deferred.py

class BranchDispatcher(Protocol):
    """Decides where the branches of one fan-out are resolved.

    ``FanOutStep`` calls ``dispatch_all`` once per fan-out with every branch. A dispatcher
    returns ``{branch.key: result}``, or raises ``FanOutBranchError`` naming the failed keys.
    A worker writes its result to the shared cache under ``branch.branch_hash``, and the
    dispatcher loads it from there. The default ``InlineDispatcher`` resolves in-process.
    """

    def dispatch_all(
        self, branches: Sequence[tuple[BranchRef, StepResult[Any]]], context: ResolveContext
    ) -> dict[str, Any]: ...

BranchRef dataclass

Everything a worker needs to re-derive and resolve one fan-out branch.

The worker imports pipeline_module, finds the FanOutStep bound to fan_out_var, looks up its own over[key] (so the value never has to serialize), calls factory(key, value) to rebuild the branch sub-DAG (so the closure never serializes either), and resolves it. branch_hash is the branch StepResult's content hash: the cache key the worker writes to and the main process polls/loads from. params are the run's InputParam values, which the worker needs to resolve to the same hash.

Source code in src/adag/deferred.py

@dataclass(frozen=True)
class BranchRef:
    """Everything a worker needs to re-derive and resolve one fan-out branch.

    The worker imports ``pipeline_module``, finds the ``FanOutStep`` bound to
    ``fan_out_var``, looks up its own ``over[key]`` (so the value never has to
    serialize), calls ``factory(key, value)`` to rebuild the branch sub-DAG (so
    the closure never serializes either), and resolves it. ``branch_hash`` is the
    branch ``StepResult``'s content hash: the cache key the worker writes to and
    the main process polls/loads from. ``params`` are the run's ``InputParam``
    values, which the worker needs to resolve to the same hash.
    """

    pipeline_module: str
    fan_out_var: str
    key: str
    branch_hash: str

Deferred

Bases: Generic[T]

A value of type T that doesn't exist yet. Resolved at execution time.

Class attribute

deferred_type: the concrete type T, auto-set when inheriting Deferred[ConcreteType].

Source code in src/adag/deferred.py

class Deferred(Generic[T]):
    """A value of type T that doesn't exist yet. Resolved at execution time.

    Class attribute:
        deferred_type: the concrete type T, auto-set when inheriting Deferred[ConcreteType].
    """

    deferred_type: type[Any] | None = None

    def __init_subclass__(cls, **kwargs: Any) -> None:
        super().__init_subclass__(**kwargs)

        for parent in getattr(cls, "__orig_bases__", []):
            origin = get_origin(parent)
            if origin is Deferred:
                # Direct Deferred[ConcreteType] base: auto-detect.
                args = get_args(parent)
                if args and isinstance(args[0], type) and "deferred_type" not in cls.__dict__:
                    cls.deferred_type = args[0]
                return

        # No direct Deferred[X] base. Check for diamond: multiple Deferred subclass bases
        # with no explicit deferred_type declaration are ambiguous.
        deferred_bases: list[Any] = []
        for parent in getattr(cls, "__orig_bases__", []):
            raw: Any = get_origin(parent) or parent
            if isinstance(raw, type) and issubclass(raw, Deferred) and raw is not Deferred:
                deferred_bases.append(raw)

        if len(deferred_bases) > 1 and "deferred_type" not in cls.__dict__:
            names = ", ".join(b.__name__ for b in deferred_bases)
            raise TypeError(
                f"'{cls.__name__}' inherits from multiple Deferred subclasses ({names}) "
                f"but does not declare 'deferred_type'. "
                f"Add 'deferred_type = <type>' to resolve the ambiguity."
            )

    def resolve(self, context: ResolveContext) -> T:
        raise NotImplementedError

    async def aresolve(self, context: ResolveContext) -> T:
        """Async resolution. The default delegates to the synchronous ``resolve``, correct for LEAF
        Deferreds whose resolution does no awaitable work (``InputParam`` reads ``context.params``,
        ``FileSource`` reads bytes). ``StepResult`` overrides this to await async step bodies and resolve
        upstreams on the event loop, so an existing sync graph runs unchanged on the async path.
        """
        return self.resolve(context)

aresolve async

aresolve(context: ResolveContext) -> T

Async resolution. The default delegates to the synchronous resolve, correct for LEAF Deferreds whose resolution does no awaitable work (InputParam reads context.params, FileSource reads bytes). StepResult overrides this to await async step bodies and resolve upstreams on the event loop, so an existing sync graph runs unchanged on the async path.

Source code in src/adag/deferred.py

async def aresolve(self, context: ResolveContext) -> T:
    """Async resolution. The default delegates to the synchronous ``resolve``, correct for LEAF
    Deferreds whose resolution does no awaitable work (``InputParam`` reads ``context.params``,
    ``FileSource`` reads bytes). ``StepResult`` overrides this to await async step bodies and resolve
    upstreams on the event loop, so an existing sync graph runs unchanged on the async path.
    """
    return self.resolve(context)

Executor

Bases: Protocol

Decides where a step runs. The graph stays the same; the executor is a property of the run, carried on ResolveContext.

run receives the step, the active context, and the already-resolved input values, and must return the step's result. The default LocalExecutor runs it in-process. A remote executor would instead dispatch the step (by its content hash) to another process or machine, wait for the result to land in the shared cache, and return it, without the graph knowing the difference.

Source code in src/adag/deferred.py

class Executor(Protocol):
    """Decides *where* a step runs. The graph stays the same; the executor is a
    property of the run, carried on ``ResolveContext``.

    ``run`` receives the step, the active context, and the already-resolved input
    values, and must return the step's result. The default ``LocalExecutor`` runs
    it in-process. A remote executor would instead dispatch the step (by its
    content hash) to another process or machine, wait for the result to land in
    the shared cache, and return it, without the graph knowing the difference.
    """

    def run(self, step: Step[Any], context: ResolveContext, inputs: dict[str, Any]) -> Any: ...

InlineDispatcher

Resolves each branch in-process, one after another. The default.

FanOutStep does not call it: it resolves inline branches itself, so it can honour max_workers. dispatch_all is here so the class meets the protocol.

Source code in src/adag/deferred.py

class InlineDispatcher:
    """Resolves each branch in-process, one after another. The default.

    ``FanOutStep`` does not call it: it resolves inline branches itself, so it can honour
    ``max_workers``. ``dispatch_all`` is here so the class meets the protocol.
    """

    def dispatch_all(
        self, branches: Sequence[tuple[BranchRef, StepResult[Any]]], context: ResolveContext
    ) -> dict[str, Any]:
        return {ref.key: sub.resolve(context) for ref, sub in branches}

    def dispatch(self, branch: BranchRef, sub: StepResult[Any], context: ResolveContext) -> Any:
        """Resolve one branch in-process. Kept from the pre-0.2.11 one-branch protocol."""
        del branch
        return sub.resolve(context)

dispatch

dispatch(
    branch: BranchRef,
    sub: StepResult[Any],
    context: ResolveContext,
) -> Any

Resolve one branch in-process. Kept from the pre-0.2.11 one-branch protocol.

Source code in src/adag/deferred.py

def dispatch(self, branch: BranchRef, sub: StepResult[Any], context: ResolveContext) -> Any:
    """Resolve one branch in-process. Kept from the pre-0.2.11 one-branch protocol."""
    del branch
    return sub.resolve(context)

InputParam dataclass

Bases: Deferred[T]

A named runtime parameter. Use with_default() for optional parameters.

text = InputParam("text", value_type=str)
threshold = InputParam.with_default("threshold", 0.5)

value_type tells the CLI how to parse --param. It is needed on a required parameter, because InputParam[int]("x") loses the int at runtime. It is not part of the cache key.

Source code in src/adag/deferred.py

@dataclass(frozen=True)
class InputParam(Deferred[T]):
    """A named runtime parameter. Use with_default() for optional parameters.

    ```python
    text = InputParam("text", value_type=str)
    threshold = InputParam.with_default("threshold", 0.5)
    ```

    ``value_type`` tells the CLI how to parse ``--param``. It is needed on a required parameter,
    because ``InputParam[int]("x")`` loses the ``int`` at runtime. It is not part of the cache key.
    """

    name: str
    default: T | None = None
    has_default: bool = False
    value_type: type[Any] | None = None

    @staticmethod
    def with_default(name: str, default: _S, value_type: type[Any] | None = None) -> InputParam[_S]:
        """Create a parameter with a default value (even if the default is None)."""
        param: Any = InputParam(name=name, default=default, has_default=True, value_type=value_type)
        return param

    def resolve(self, context: ResolveContext) -> T:
        if self.name in context.params and context.params[self.name] is not None:
            return context.params[self.name]
        if self.has_default:
            default_value: Any = self.default
            return default_value
        raise ValueError(f"Missing required parameter: {self.name}")

with_default staticmethod

with_default(
    name: str,
    default: _S,
    value_type: type[Any] | None = None,
) -> InputParam[_S]

Create a parameter with a default value (even if the default is None).

Source code in src/adag/deferred.py

@staticmethod
def with_default(name: str, default: _S, value_type: type[Any] | None = None) -> InputParam[_S]:
    """Create a parameter with a default value (even if the default is None)."""
    param: Any = InputParam(name=name, default=default, has_default=True, value_type=value_type)
    return param

LocalAsyncExecutor

Awaits each step's async body in-process, the default async executor. The mirror of LocalExecutor for the aresolve path.

Source code in src/adag/deferred.py

class LocalAsyncExecutor:
    """Awaits each step's async body in-process, the default async executor. The mirror of
    ``LocalExecutor`` for the ``aresolve`` path.
    """

    async def run(self, step: Step[Any], context: ResolveContext, inputs: dict[str, Any]) -> Any:
        return await step.aexecute(context, **inputs)

LocalExecutor

Runs each step in this process. The default.

Source code in src/adag/deferred.py

class LocalExecutor:
    """Runs each step in this process. The default."""

    def run(self, step: Step[Any], context: ResolveContext, inputs: dict[str, Any]) -> Any:
        return step.execute(context, **inputs)

NullObserver

Silent observer: every hook is a no-op. Use it to turn the per-step logging off. Concrete observers inherit this and override only the hooks they care about, so new hooks never break an existing observer.

Source code in src/adag/deferred.py

class NullObserver:
    """Silent observer: every hook is a no-op. Use it to turn the per-step
    logging off. Concrete observers inherit this and override only the hooks
    they care about, so new hooks never break an existing observer.
    """

    def on_step_start(self, step: str) -> None:
        pass

    def on_step_cached(self, step: str) -> None:
        pass

    def on_step_finished(self, step: str, seconds: float) -> None:
        pass

    def on_fan_out_branch(self, key: str) -> None:
        pass

    def on_fan_out_branch_failed(self, key: str, error: str) -> None:
        pass

    def on_event(self, event: str, fields: dict[str, Any]) -> None:
        pass

    def on_embedding(self, vector: list[float]) -> None:
        """Deprecated no-op, kept for code that calls it. Use a typed hook (docs/observers.md)."""
        _warn_domain_hook("on_embedding")

    def on_model_refused(self, model: str) -> None:
        """Deprecated no-op, kept for code that calls it. Use a typed hook (docs/observers.md)."""
        _warn_domain_hook("on_model_refused")

on_embedding

on_embedding(vector: list[float]) -> None

Deprecated no-op, kept for code that calls it. Use a typed hook (docs/observers.md).

Source code in src/adag/deferred.py

def on_embedding(self, vector: list[float]) -> None:
    """Deprecated no-op, kept for code that calls it. Use a typed hook (docs/observers.md)."""
    _warn_domain_hook("on_embedding")

on_model_refused

on_model_refused(model: str) -> None

Deprecated no-op, kept for code that calls it. Use a typed hook (docs/observers.md).

Source code in src/adag/deferred.py

def on_model_refused(self, model: str) -> None:
    """Deprecated no-op, kept for code that calls it. Use a typed hook (docs/observers.md)."""
    _warn_domain_hook("on_model_refused")

PipelineObserver

Bases: Protocol

Hooks a pipeline run calls so you can watch it: logs, timing, progress.

StepResult.resolve calls on_step_start before it resolves a step's inputs, and on_step_cached or on_step_finished after. A child's hooks fall between its parent's start and finish. RunTreeObserver builds its tree from that order.

Subclass NullObserver and override the hooks you need. For a value of your own (a score, a vector), declare a typed hook: see "Typed hooks for your own values" in docs/observers.md.

Source code in src/adag/deferred.py

class PipelineObserver(Protocol):
    """Hooks a pipeline run calls so you can watch it: logs, timing, progress.

    ``StepResult.resolve`` calls ``on_step_start`` before it resolves a step's inputs, and
    ``on_step_cached`` or ``on_step_finished`` after. A child's hooks fall between its parent's
    start and finish. ``RunTreeObserver`` builds its tree from that order.

    Subclass ``NullObserver`` and override the hooks you need. For a value of your own (a score,
    a vector), declare a typed hook: see "Typed hooks for your own values" in docs/observers.md.
    """

    def on_step_start(self, step: str) -> None: ...

    def on_step_cached(self, step: str) -> None: ...

    def on_step_finished(self, step: str, seconds: float) -> None: ...

    def on_fan_out_branch(self, key: str) -> None: ...

    def on_fan_out_branch_failed(self, key: str, error: str) -> None: ...

    def on_event(self, event: str, fields: dict[str, Any]) -> None: ...

Progress dataclass

Progress report from a resolving Deferred.

Source code in src/adag/deferred.py

@dataclass(frozen=True)
class Progress:
    """Progress report from a resolving Deferred."""

    step: str
    message: str
    fraction: float | None = None
    data: Any = None

ResolveContext dataclass

The state of one run: where results are cached, the params, the observer and the memo.

Build one per run and pass it to every resolve, aresolve or aresolve_all call.

• cache_dir: files for step results. store replaces it with any CacheStore. With neither, nothing is cached.
• params: the InputParam values of the run.
• observer: NullObserver by default here. Pipeline.execute passes LoggingObserver unless you give another.
• The memo: each node resolves once per context, so every downstream step gets the same object. Threads or tasks that reach one node at the same time run its body once.

Source code in src/adag/deferred.py

@dataclass(frozen=True)
class ResolveContext:
    """The state of one run: where results are cached, the params, the observer and the memo.

    Build one per run and pass it to every ``resolve``, ``aresolve`` or ``aresolve_all`` call.

    - ``cache_dir``: files for step results. ``store`` replaces it with any ``CacheStore``. With
      neither, nothing is cached.
    - ``params``: the ``InputParam`` values of the run.
    - ``observer``: ``NullObserver`` by default here. ``Pipeline.execute`` passes
      ``LoggingObserver`` unless you give another.
    - The memo: each node resolves once per context, so every downstream step gets the same
      object. Threads or tasks that reach one node at the same time run its body once.
    """

    cache_dir: Path | None = None
    params: dict[str, Any] = field(default_factory=_empty_params)
    on_progress: Callable[[Progress], None] | None = None
    observer: PipelineObserver = field(default_factory=NullObserver)
    executor: Executor = field(default_factory=LocalExecutor)
    async_executor: AsyncExecutor = field(default_factory=LocalAsyncExecutor)
    dispatcher: BranchDispatcher = field(default_factory=InlineDispatcher)
    # Last, so positional ResolveContext(cache_dir, params, on_progress, ...) from 0.1 still binds.
    store: CacheStore | None = None
    # The memo: ``id(node) -> (node, value)``. Without it a step that returns a non-serializable
    # object (a model handle) would run once per consumer, and each consumer would get a
    # different instance.
    _resolved: dict[int, Any] = field(default_factory=_empty_resolved, repr=False, compare=False)
    # While one thread runs a node's body, its id maps to an Event that other threads wait on.
    # The owner removes the entry and sets the Event in a ``finally``, so a body that raises
    # releases its waiters, which then resolve the node again.
    _inflight: dict[int, Event] = field(default_factory=_empty_inflight, repr=False, compare=False)
    # Guards every read and write of the two dicts above. A level of nodes resolves in a thread
    # pool, and the free-threaded build gives no atomicity for free.
    _lock: Lock = field(default_factory=Lock, repr=False, compare=False)
    # The in-flight registry for ``aresolve``: a future per node, so gathered tasks wait on the
    # loop instead of blocking it on a thread Event.
    _ainflight: dict[int, asyncio.Future[None]] = field(
        default_factory=_empty_ainflight, repr=False, compare=False
    )

    def result_store(self) -> CacheStore | None:
        """Where step results are cached: ``store``, else files under ``cache_dir``, else None."""
        if self.store is not None:
            return self.store
        if self.cache_dir is None:
            return None
        from adag.cache_store import FileSystemCacheStore  # late: cache_store imports io

        return FileSystemCacheStore(self.cache_dir)

result_store

result_store() -> CacheStore | None

Where step results are cached: store, else files under cache_dir, else None.

Source code in src/adag/deferred.py

def result_store(self) -> CacheStore | None:
    """Where step results are cached: ``store``, else files under ``cache_dir``, else None."""
    if self.store is not None:
        return self.store
    if self.cache_dir is None:
        return None
    from adag.cache_store import FileSystemCacheStore  # late: cache_store imports io

    return FileSystemCacheStore(self.cache_dir)

AdagDeprecationWarning

Bases: FutureWarning

A deprecated adag name or call form. Shown by default; removed in adag 1.0.

Source code in src/adag/deprecation.py

class AdagDeprecationWarning(FutureWarning):
    """A deprecated adag name or call form. Shown by default; removed in adag 1.0."""

DirSource dataclass

Bases: Deferred[str]

A folder that acts as a leaf node in the step graph.

Hashed by the SHA-256 of a deterministic stream of (relative_path, file_hash) pairs across every file matching pattern inside the folder. If the folder doesn't exist, the hash returns None, blocking any downstream step that depends on it.

pattern defaults to "*.jsonl", which is not recursive. Pass another glob (for example "**/*.json") for nested folders.

Source code in src/adag/dir_source.py

@dataclass(frozen=True)
class DirSource(Deferred[str]):
    """A folder that acts as a leaf node in the step graph.

    Hashed by the SHA-256 of a deterministic stream of
    ``(relative_path, file_hash)`` pairs across every file matching
    ``pattern`` inside the folder. If the folder doesn't
    exist, the hash returns ``None``, blocking any downstream step that
    depends on it.

    ``pattern`` defaults to ``"*.jsonl"``, which is not recursive. Pass another glob (for
    example ``"**/*.json"``) for nested folders.
    """

    path: str
    pattern: str = "*.jsonl"

    @property
    def _path(self) -> Path:
        return Path(self.path)

    def _list_files(self) -> list[Path]:
        """All matching files, in stable sorted order by relative path."""
        if not self._path.exists() or not self._path.is_dir():
            return []
        return sorted(
            self._path.glob(self.pattern), key=lambda p: p.relative_to(self._path).as_posix()
        )

    def content_hash(self) -> str | None:
        """SHA-256 over ``(relative_path, file_hash)`` for each file.

        First 12 hex chars to match ``FileSource``. Returns ``None`` if
        the directory doesn't exist (treats as blocked).
        """
        if not self._path.exists() or not self._path.is_dir():
            return None
        h = hashlib.sha256()
        for p in self._list_files():
            rel = p.relative_to(self._path).as_posix()
            h.update(rel.encode())
            h.update(b"|")
            fh = file_hash(p)
            h.update(fh.encode())
            h.update(b"\n")
        return h.hexdigest()[:12]

    def resolve(self, context: ResolveContext) -> str:
        """Return the folder path. Steps load files themselves."""
        del context
        return str(self._path)

content_hash

content_hash() -> str | None

SHA-256 over (relative_path, file_hash) for each file.

First 12 hex chars to match FileSource. Returns None if the directory doesn't exist (treats as blocked).

Source code in src/adag/dir_source.py

def content_hash(self) -> str | None:
    """SHA-256 over ``(relative_path, file_hash)`` for each file.

    First 12 hex chars to match ``FileSource``. Returns ``None`` if
    the directory doesn't exist (treats as blocked).
    """
    if not self._path.exists() or not self._path.is_dir():
        return None
    h = hashlib.sha256()
    for p in self._list_files():
        rel = p.relative_to(self._path).as_posix()
        h.update(rel.encode())
        h.update(b"|")
        fh = file_hash(p)
        h.update(fh.encode())
        h.update(b"\n")
    return h.hexdigest()[:12]

resolve

resolve(context: ResolveContext) -> str

Return the folder path. Steps load files themselves.

Source code in src/adag/dir_source.py

def resolve(self, context: ResolveContext) -> str:
    """Return the folder path. Steps load files themselves."""
    del context
    return str(self._path)

BranchPolicy dataclass

How hard to try a branch and how patiently to wait for it.

timeout_seconds counts from when the job is RUNNING. queue_timeout_seconds counts from submit while the job is PENDING; a batch queue can hold a job for hours. result_grace_seconds is how long to wait for the result to appear after a job SUCCEEDED, because a network filesystem can show a new file late.

Source code in src/adag/dispatch.py

@dataclass(frozen=True)
class BranchPolicy:
    """How hard to try a branch and how patiently to wait for it.

    ``timeout_seconds`` counts from when the job is RUNNING. ``queue_timeout_seconds`` counts
    from submit while the job is PENDING; a batch queue can hold a job for hours.
    ``result_grace_seconds`` is how long to wait for the result to appear after a job
    SUCCEEDED, because a network filesystem can show a new file late.
    """

    max_retries: int = 0
    timeout_seconds: float | None = None
    poll_interval_seconds: float = 5.0
    retry_delay_seconds: float = 0.0
    result_grace_seconds: float = 5.0
    # Added in 0.2.11. Kept after the first-release fields so BranchPolicy(1, 60, 2.0) still means
    # (max_retries, timeout_seconds, poll_interval_seconds).
    queue_timeout_seconds: float | None = None

SubprocessDispatcher

Bases: RemoteDispatcher

Resolve each branch in a local OS process via adag resolve-branch.

RemoteDispatcher with LocalJobBackend. Runs at most max_concurrent processes at once (default: the number of CPUs). Workers share results through cache_dir, or the store at store_url. The pipeline must have one of them.

Source code in src/adag/dispatch.py

class SubprocessDispatcher(RemoteDispatcher):
    """Resolve each branch in a local OS process via ``adag resolve-branch``.

    ``RemoteDispatcher`` with ``LocalJobBackend``. Runs at most ``max_concurrent`` processes at
    once (default: the number of CPUs). Workers share results through ``cache_dir``, or the
    store at ``store_url``. The pipeline must have one of them.
    """

    def __init__(
        self,
        policy: BranchPolicy | None = None,
        max_concurrent: int | None = None,
        store_url: str | None = None,
    ) -> None:
        super().__init__(
            LocalJobBackend(),
            policy or BranchPolicy(poll_interval_seconds=0.05),
            max_concurrent=max_concurrent or os.cpu_count() or 1,
            store_url=store_url,
        )

FanOutBranchError

Bases: RuntimeError

One or more distributed fan-out branches failed.

Succeeded branches' results are already in the shared cache (written by their workers, keyed by content hash), so a rerun re-dispatches only the failed keys.

Source code in src/adag/fan_out.py

class FanOutBranchError(RuntimeError):
    """One or more distributed fan-out branches failed.

    Succeeded branches' results are already in the shared cache (written by their
    workers, keyed by content hash), so a rerun re-dispatches only the failed keys.
    """

    def __init__(self, succeeded: list[str], failed: dict[str, str]) -> None:
        self.succeeded = succeeded
        self.failed = failed
        detail = "; ".join(f"{key}: {error}" for key, error in failed.items())
        super().__init__(
            f"{len(failed)} fan-out branch(es) failed ({detail}); {len(succeeded)} succeeded."
        )

FanOutStep dataclass

Bases: Step[dict[K, V]], Generic[K, V]

Apply a sub-pipeline factory to each value in over; collect into a dict.

Fields

over: dict whose keys identify each fan-out branch and whose values are passed verbatim to factory for that branch. factory: callable (key, value) -> StepResult[V] building the sub-DAG for one branch. Called once per key during execute; each returned StepResult is resolved under the same ResolveContext as the parent so caches are shared. max_workers: run branches in-process on this many threads. Ignored by a dispatcher. instance_label: overrides the visualization label. pipeline_module: file path or module name a dispatch worker imports. fan_out_var: module-level variable that holds this fan-out's StepResult.

Source code in src/adag/fan_out.py

@dataclass(frozen=True, eq=False)
class FanOutStep(Step[dict[K, V]], Generic[K, V]):
    """Apply a sub-pipeline factory to each value in ``over``; collect into a dict.

    Fields:
        over: dict whose keys identify each fan-out branch and whose values
            are passed verbatim to ``factory`` for that branch.
        factory: callable ``(key, value) -> StepResult[V]`` building the
            sub-DAG for one branch. Called once per key during ``execute``;
            each returned ``StepResult`` is resolved under the same
            ``ResolveContext`` as the parent so caches are shared.
        max_workers: run branches in-process on this many threads. Ignored by a dispatcher.
        instance_label: overrides the visualization label.
        pipeline_module: file path or module name a dispatch worker imports.
        fan_out_var: module-level variable that holds this fan-out's ``StepResult``.
    """

    over: dict[K, Any]
    factory: Callable[[K, Any], StepResult[V]]
    # Concurrency of the in-process (InlineDispatcher) path. None (default) keeps the
    # historical sequential branch-by-branch resolution. An int resolves branches in a
    # ThreadPoolExecutor of that size, a win when each branch is I/O-bound (an LLM/HTTP
    # call). Shared upstreams stay safe under it: ResolveContext's in-flight dedup runs a
    # shared node's body once even when many branch workers reach it at the same instant.
    # Ignored by a non-inline dispatcher, which has its own concurrency.
    max_workers: int | None = field(default=None)
    # Optional human-readable label for visualization. Mirrors the
    # ClassVar on Step but as a per-instance override so a notebook can
    # name two FanOuts differently without subclassing.
    instance_label: str = field(default="")
    # For distributed dispatch (e.g. SubprocessDispatcher): the file path or the module
    # name (pkg.pipeline) of the module that defines this fan-out and the module-level variable its
    # StepResult is bound to, so a worker can re-import and re-derive a branch
    # via factory(key, over[key]). Empty for in-process runs (InlineDispatcher),
    # which never need them. Not part of content_hash (it ignores fields).
    pipeline_module: str = field(default="")
    fan_out_var: str = field(default="")

    def execute(self, context: ResolveContext, /, **kwargs: Any) -> dict[K, V]:
        """Resolve one sub-DAG per key in ``over``.

        In-process (InlineDispatcher, the default) each branch resolves under the parent
        context: branch-by-branch when ``max_workers`` is None (identical to the pre-dispatch
        behaviour) or concurrently in a thread pool when it is set. With a real dispatcher,
        branches are dispatched to workers and rejoined through the shared cache by content hash.
        """
        del kwargs  # FanOutStep does not take wired inputs, closure over factory does.
        if isinstance(context.dispatcher, InlineDispatcher):
            if self.max_workers is None:
                return self._resolve_branches_sequentially(context)
            return self._resolve_branches_concurrently(context, self.max_workers)
        return self._dispatch_branches(context)

    async def aexecute(self, context: ResolveContext, /, **kwargs: Any) -> dict[K, V]:
        """Async entry: run ``execute`` in a worker thread so the event loop is not blocked.

        Branches resolve on the sync path inside that thread, the same as under ``execute``.
        ``asyncio.to_thread`` copies the current context, so the observer reaches the branches.
        """
        return await asyncio.to_thread(self.execute, context, **kwargs)

    def _resolve_branches_sequentially(self, context: ResolveContext) -> dict[K, V]:
        """Resolve each branch under the parent context, one after another."""
        results: dict[K, V] = {}
        # Hold every branch StepResult alive for the whole fan-out: the per-execute memo keys by
        # id(), so a branch freed before the next is built would let CPython reuse its address and
        # the next branch would hit its stale memo entry.
        branches: list[StepResult[V]] = []
        for key, value in self.over.items():
            # Tell the observer which branch is next so it can label this branch's
            # subtree by key (a no-op for every observer except RunTreeObserver).
            context.observer.on_fan_out_branch(str(key))
            sub = self._branch(key, value)
            branches.append(sub)
            results[key] = sub.resolve(context)
        return results

    def _resolve_branches_concurrently(
        self, context: ResolveContext, max_workers: int
    ) -> dict[K, V]:
        """Resolve every branch in a thread pool, each worker inside a COPIED context so ambient
        ContextVars (the usage observer, ``current_observer``) propagate across the thread boundary.

        A branch raising propagates out of ``future.result()``, the same as on the sequential path,
        where a branch raising aborts ``execute``. It is not collected, as it is on the dispatcher path. The
        pool's ``__exit__`` waits for the in-flight branches before the exception unwinds.
        """
        if not self.over:
            return {}
        prepared: list[tuple[K, StepResult[V]]] = []
        for key, value in self.over.items():
            context.observer.on_fan_out_branch(str(key))
            prepared.append((key, self._branch(key, value)))
        results: dict[K, V] = {}
        with ThreadPoolExecutor(max_workers=max_workers) as pool:
            futures: dict[Future[V], K] = {
                pool.submit(copy_context().run, sub.resolve, context): key for key, sub in prepared
            }
            for future in as_completed(futures):
                results[futures[future]] = future.result()
        return {key: results[key] for key, _ in prepared}  # keep the order of ``over``

    def _branch(self, key: K, value: Any) -> StepResult[V]:
        """Build one branch's sub-DAG via the factory, validating it is a resolvable Deferred."""
        sub = self.factory(key, value)
        if not isinstance(sub, StepResult):  # pyright: ignore[reportUnnecessaryIsInstance]  (factory is user code)
            raise TypeError(
                f"FanOutStep factory must return a StepResult/Deferred; got {type(sub).__name__} for key {key!r}."
            )
        return sub

    def _dispatch_branches(self, context: ResolveContext) -> dict[K, V]:
        """Distribute branches via ``context.dispatcher`` and rejoin by content hash.

        Pass 1 re-derives each branch (runs the factory), validates it can cross the wire, and
        builds a ``BranchRef``. Pass 2 hands all branches to ``dispatcher.dispatch_all`` in one
        call, so the dispatcher controls concurrency and polls every job in one loop.
        """
        if not self.pipeline_module or not self.fan_out_var:
            raise ValueError(
                "Distributed FanOutStep requires pipeline_module and fan_out_var so a worker can re-derive branches; "
                "set them when building the fan-out for distributed dispatch."
            )

        seen: dict[str, K] = {}
        for key in self.over:
            if str(key) in seen:
                raise ValueError(
                    f"FanOutStep keys {seen[str(key)]!r} and {key!r} have the same text "
                    f"{str(key)!r}; a worker finds its branch by that text, so keys must differ."
                )
            seen[str(key)] = key

        prepared: list[tuple[K, StepResult[V], BranchRef]] = []
        for key, value in self.over.items():
            context.observer.on_fan_out_branch(str(key))
            sub = self.factory(key, value)
            if not isinstance(sub, StepResult):  # pyright: ignore[reportUnnecessaryIsInstance]  (factory is user code)
                raise TypeError(
                    f"FanOutStep factory must return a StepResult; got {type(sub).__name__} for key {key!r}."
                )
            if sub.step.result_model() is None:
                raise TypeError(
                    f"Distributed FanOutStep branch {key!r} ({type(sub.step).__name__}) has no result_type; "
                    "distributed branches must terminate in a step whose result is a BaseModel (so it can cross the cache)."
                )
            branch_hash = sub.content_hash(params=context.params)
            if branch_hash is None:
                raise ValueError(
                    f"FanOutStep branch {key!r} has no content hash (blocked input); cannot dispatch."
                )
            prepared.append(
                (key, sub, BranchRef(self.pipeline_module, self.fan_out_var, str(key), branch_hash))
            )

        if not prepared:
            return {}

        dispatch_all = getattr(context.dispatcher, "dispatch_all", None)
        if dispatch_all is None:
            return self._dispatch_one_by_one(context, prepared)
        by_key = dispatch_all([(ref, sub) for _, sub, ref in prepared], context)
        return {key: by_key[ref.key] for key, _, ref in prepared}  # keep the order of ``over``

    def _dispatch_one_by_one(
        self, context: ResolveContext, prepared: list[tuple[K, StepResult[V], BranchRef]]
    ) -> dict[K, V]:
        """Run a dispatcher that has only the pre-0.2.11 ``dispatch(branch, sub, context)``.

        Runs one thread per branch. Deprecated: implement ``dispatch_all``.
        """
        warnings.warn(
            f"{type(context.dispatcher).__name__} has only dispatch(); implement "
            "dispatch_all(branches, context). dispatch() support is removed in adag 1.0.",
            AdagDeprecationWarning,
            stacklevel=2,
        )
        dispatch: Any = getattr(context.dispatcher, "dispatch")  # noqa: B009
        results: dict[K, V] = {}
        failed: dict[str, str] = {}
        with ThreadPoolExecutor(max_workers=len(prepared)) as pool:
            futures = {
                pool.submit(copy_context().run, dispatch, ref, sub, context): key
                for key, sub, ref in prepared
            }
            for future in as_completed(futures):
                key = futures[future]
                try:
                    results[key] = future.result()
                except Exception as exc:
                    failed[str(key)] = str(exc)
                    context.observer.on_fan_out_branch_failed(str(key), str(exc))
        if failed:
            raise FanOutBranchError(succeeded=sorted(str(key) for key in results), failed=failed)
        return {key: results[key] for key, _, _ in prepared}

    def content_hash(self) -> str | None:
        """Identity = class + each branch's sub-DAG content_hash.

        ``factory`` itself can't be hashed (closures aren't stably
        hashable), so we materialize each branch once and fold the
        resulting ``StepResult.content_hash()`` into our own digest.
        Keys are included so reordering / renaming branches invalidates
        the cache; values only contribute via the sub-DAG hash they
        produce. That is what matters for correctness.

        Returns None when a branch's hash is blocked (e.g. unsatisfied
        upstream FileSource), so the parent pipeline behaves like any
        other unsatisfied step.
        """
        h = hashlib.sha256()
        h.update(b"FANOUT:" + type(self).__name__.encode())
        for key in sorted(self.over, key=repr):
            try:
                sub = self.factory(key, self.over[key])
            except Exception as exc:
                logger.warning(
                    "fan_out.factory_raised",
                    step=type(self).__name__,
                    key=repr(key),
                    error_type=type(exc).__name__,
                    error=str(exc)[:200],
                )
                return None
            sub_hash = sub.content_hash() if isinstance(sub, StepResult) else None  # pyright: ignore[reportUnnecessaryIsInstance]
            if sub_hash is None:
                return None
            h.update(b"|K:" + repr(key).encode() + b"|H:" + sub_hash.encode())
        return h.hexdigest()[:12]

execute

execute(
    context: ResolveContext, /, **kwargs: Any
) -> dict[K, V]

Resolve one sub-DAG per key in over.

In-process (InlineDispatcher, the default) each branch resolves under the parent context: branch-by-branch when max_workers is None (identical to the pre-dispatch behaviour) or concurrently in a thread pool when it is set. With a real dispatcher, branches are dispatched to workers and rejoined through the shared cache by content hash.

Source code in src/adag/fan_out.py

def execute(self, context: ResolveContext, /, **kwargs: Any) -> dict[K, V]:
    """Resolve one sub-DAG per key in ``over``.

    In-process (InlineDispatcher, the default) each branch resolves under the parent
    context: branch-by-branch when ``max_workers`` is None (identical to the pre-dispatch
    behaviour) or concurrently in a thread pool when it is set. With a real dispatcher,
    branches are dispatched to workers and rejoined through the shared cache by content hash.
    """
    del kwargs  # FanOutStep does not take wired inputs, closure over factory does.
    if isinstance(context.dispatcher, InlineDispatcher):
        if self.max_workers is None:
            return self._resolve_branches_sequentially(context)
        return self._resolve_branches_concurrently(context, self.max_workers)
    return self._dispatch_branches(context)

aexecute async

aexecute(
    context: ResolveContext, /, **kwargs: Any
) -> dict[K, V]

Async entry: run execute in a worker thread so the event loop is not blocked.

Branches resolve on the sync path inside that thread, the same as under execute. asyncio.to_thread copies the current context, so the observer reaches the branches.

Source code in src/adag/fan_out.py

async def aexecute(self, context: ResolveContext, /, **kwargs: Any) -> dict[K, V]:
    """Async entry: run ``execute`` in a worker thread so the event loop is not blocked.

    Branches resolve on the sync path inside that thread, the same as under ``execute``.
    ``asyncio.to_thread`` copies the current context, so the observer reaches the branches.
    """
    return await asyncio.to_thread(self.execute, context, **kwargs)

content_hash

content_hash() -> str | None

Identity = class + each branch's sub-DAG content_hash.

factory itself can't be hashed (closures aren't stably hashable), so we materialize each branch once and fold the resulting StepResult.content_hash() into our own digest. Keys are included so reordering / renaming branches invalidates the cache; values only contribute via the sub-DAG hash they produce. That is what matters for correctness.

Returns None when a branch's hash is blocked (e.g. unsatisfied upstream FileSource), so the parent pipeline behaves like any other unsatisfied step.

Source code in src/adag/fan_out.py

def content_hash(self) -> str | None:
    """Identity = class + each branch's sub-DAG content_hash.

    ``factory`` itself can't be hashed (closures aren't stably
    hashable), so we materialize each branch once and fold the
    resulting ``StepResult.content_hash()`` into our own digest.
    Keys are included so reordering / renaming branches invalidates
    the cache; values only contribute via the sub-DAG hash they
    produce. That is what matters for correctness.

    Returns None when a branch's hash is blocked (e.g. unsatisfied
    upstream FileSource), so the parent pipeline behaves like any
    other unsatisfied step.
    """
    h = hashlib.sha256()
    h.update(b"FANOUT:" + type(self).__name__.encode())
    for key in sorted(self.over, key=repr):
        try:
            sub = self.factory(key, self.over[key])
        except Exception as exc:
            logger.warning(
                "fan_out.factory_raised",
                step=type(self).__name__,
                key=repr(key),
                error_type=type(exc).__name__,
                error=str(exc)[:200],
            )
            return None
        sub_hash = sub.content_hash() if isinstance(sub, StepResult) else None  # pyright: ignore[reportUnnecessaryIsInstance]
        if sub_hash is None:
            return None
        h.update(b"|K:" + repr(key).encode() + b"|H:" + sub_hash.encode())
    return h.hexdigest()[:12]

UnhashableValueError

Bases: TypeError

A value has a type the cache key cannot encode.

Source code in src/adag/hashing.py

class UnhashableValueError(TypeError):
    """A value has a type the cache key cannot encode."""

FileSource dataclass

Bases: Deferred[str]

Reads the file's full text content into a single string.

Source code in src/adag/io.py

@dataclass(frozen=True)
class FileSource(Deferred[str]):
    """Reads the file's full text content into a single string."""

    path: str

    def content_hash(self) -> str | None:
        return _existing_path_hash(self.path, type(self).__name__)

    def resolve(self, context: ResolveContext) -> str:
        del context  # leaf source: no cache_dir or params consulted
        return Path(self.path).read_text()

JSONLSource dataclass

Bases: Deferred[list[T]], Generic[T]

Reads a JSONL file and validates each non-blank line against record_type.

record_type is required (not optional). The Pydantic class is what gives this leaf its precise output type. Making it optional would force a list[T] | list[Any] return signature and erase the typing benefit. If you want untyped lines, use LinesSource and call json.loads in a downstream Step.

Usage::

source = JSONLSource(path="data/reports.jsonl", record_type=ReportEntry)
records: list[ReportEntry] = source.resolve(context)

A line that fails to parse or validate raises ValueError with the file path and line number: the original error is chained via raise … from. Blank lines are skipped silently.

Source code in src/adag/io.py

@dataclass(frozen=True)
class JSONLSource(Deferred[list[T]], Generic[T]):
    """Reads a JSONL file and validates each non-blank line against ``record_type``.

    ``record_type`` is required (not optional). The Pydantic class is what
    gives this leaf its precise output type. Making it optional would force
    a ``list[T] | list[Any]`` return signature and erase the typing benefit.
    If you want untyped lines, use ``LinesSource`` and call ``json.loads`` in
    a downstream Step.

    Usage::

        source = JSONLSource(path="data/reports.jsonl", record_type=ReportEntry)
        records: list[ReportEntry] = source.resolve(context)

    A line that fails to parse or validate raises ``ValueError`` with the file
    path and line number: the original error is chained via ``raise … from``.
    Blank lines are skipped silently.
    """

    path: str
    record_type: type[T]

    def content_hash(self) -> str | None:
        return _existing_path_hash(self.path, type(self).__name__)

    def resolve(self, context: ResolveContext) -> list[T]:
        del context  # leaf source: no cache_dir or params consulted
        records: list[T] = []
        for line_number, line in enumerate(Path(self.path).read_text().splitlines(), start=1):
            stripped = line.strip()
            if not stripped:
                continue
            try:
                records.append(self.record_type.model_validate_json(stripped))
            except Exception as error:
                raise ValueError(f"{self.path}:{line_number}: {error}") from error
        return records

LinesSource dataclass

Bases: Deferred[list[str]]

Reads the file as a list of lines (raw, line endings stripped).

Blank lines are preserved: this is the literal contents of the file split on newlines, equivalent to Path(path).read_text().splitlines(). Use JSONLSource if you want blanks skipped and each line parsed.

Source code in src/adag/io.py

@dataclass(frozen=True)
class LinesSource(Deferred[list[str]]):
    """Reads the file as a list of lines (raw, line endings stripped).

    Blank lines are preserved: this is the literal contents of the file split
    on newlines, equivalent to ``Path(path).read_text().splitlines()``. Use
    ``JSONLSource`` if you want blanks skipped and each line parsed.
    """

    path: str

    def content_hash(self) -> str | None:
        return _existing_path_hash(self.path, type(self).__name__)

    def resolve(self, context: ResolveContext) -> list[str]:
        del context  # leaf source: no cache_dir or params consulted
        return Path(self.path).read_text().splitlines()

ItemCache

Bases: Generic[R]

Per-item cache for expensive batch operations.

Items are stored as individual JSON values keyed by a content hash. This survives parameter changes: only the item's own identity matters.

Usage::

cache = ItemCache[NoteSchema](cache_dir, "note_generation", NoteSchema)

keys = [cache.key(item.code, style_hash, model) for item in items]
cached, uncached_indices = cache.partition(keys)

for index in uncached_indices:
    result = expensive_llm_call(items[index])
    cache.save(keys[index], result)

all_results = cache.collect(keys)

Pass store= to back the cache with something other than the local filesystem (e.g. a RedisCacheStore for a deployed multi-instance service). Default = FileSystemCacheStore rooted at <cache_dir>/.items (the historical layout).

Source code in src/adag/item_cache.py

class ItemCache(Generic[R]):
    """Per-item cache for expensive batch operations.

    Items are stored as individual JSON values keyed by a content hash.
    This survives parameter changes: only the item's own identity matters.

    Usage::

        cache = ItemCache[NoteSchema](cache_dir, "note_generation", NoteSchema)

        keys = [cache.key(item.code, style_hash, model) for item in items]
        cached, uncached_indices = cache.partition(keys)

        for index in uncached_indices:
            result = expensive_llm_call(items[index])
            cache.save(keys[index], result)

        all_results = cache.collect(keys)

    Pass ``store=`` to back the cache with something other than the local filesystem (e.g. a
    ``RedisCacheStore`` for a deployed multi-instance service). Default = ``FileSystemCacheStore`` rooted at
    ``<cache_dir>/.items`` (the historical layout).
    """

    def __init__(
        self, cache_dir: Path, namespace: str, schema: type[R], *, store: CacheStore | None = None
    ) -> None:
        self.cache_dir = cache_dir
        self.namespace = namespace
        self.schema: type[R] = schema
        self.store: CacheStore = (
            store if store is not None else FileSystemCacheStore(Path(cache_dir).joinpath(".items"))
        )

    # Keys in the store are namespaced + suffixed so result files and in-flight markers coexist and the
    # default filesystem layout (<cache_dir>/.items/<namespace>/<key>.json) is preserved exactly.
    def _item_key(self, item_key: str) -> str:
        return f"{self.namespace}/{item_key}.json"

    def _inflight_key(self, item_key: str) -> str:
        return f"{self.namespace}/{item_key}.inflight"

    def key(self, *parts: str) -> str:
        """Compute a cache key from the item's identity parts."""
        h = hashlib.sha256()
        for part in parts:
            h.update(part.encode())
        return h.hexdigest()[:16]

    def _schema_version(self) -> str:
        """Stable digest of the schema's JSON-schema representation.

        Encodes the field names, types, requireds, and nesting. If the
        schema changes (field added, renamed, retyped) the digest changes,
        old caches become stale, and partition() treats them as misses.
        """
        spec = self.schema.model_json_schema()
        return hashlib.sha256(json.dumps(spec, sort_keys=True).encode()).hexdigest()[:12]

    def exists(self, item_key: str) -> bool:
        """Check if an item is cached."""
        return self.store.exists(self._item_key(item_key))

    # Alias: "has" reads more naturally for cache operations.
    has = exists

    def list_keys(self) -> set[str]:
        """Return all cached item keys."""
        prefix = f"{self.namespace}/"
        return {
            k[len(prefix) : -len(".json")]
            for k in self.store.iter_keys(prefix)
            if k.endswith(".json")
        }

    # ------------------------------------------------------------------
    # In-flight WAL
    # ------------------------------------------------------------------

    def mark_inflight(self, item_key: str) -> None:
        """Record that processing for ``item_key`` has started.

        Writes a ``<key>.inflight`` sentinel. If the worker crashes before
        ``save()``, this sentinel persists and can be surfaced on next
        resume to flag work-in-progress items that were lost. Call
        ``clear_inflight()`` after a successful save.
        """
        self.store.set(self._inflight_key(item_key), "")

    def clear_inflight(self, item_key: str) -> None:
        """Remove the in-flight sentinel after a successful save."""
        self.store.delete(self._inflight_key(item_key))

    def inflight_keys(self) -> set[str]:
        """All currently-marked in-flight item keys (typically from a crashed run)."""
        prefix = f"{self.namespace}/"
        return {
            k[len(prefix) : -len(".inflight")]
            for k in self.store.iter_keys(prefix)
            if k.endswith(".inflight")
        }

    # ------------------------------------------------------------------
    # Save / load with schema-version wrapping
    # ------------------------------------------------------------------

    def save(self, item_key: str, result: R) -> None:
        """Save a single item result with a schema-version stamp.

        Stamping the schema version on every write lets a future
        ``load()`` detect when the cached payload was produced under a
        different schema and refuse to deserialize it. Old, pre-stamp
        caches still load via the legacy path below for backwards
        compatibility.
        """
        wrapper: dict[str, Any] = {
            _SCHEMA_VERSION_FIELD: self._schema_version(),
            _DATA_FIELD: result.model_dump(mode="json"),
        }
        self.store.set(self._item_key(item_key), json.dumps(wrapper, indent=2))
        self.clear_inflight(item_key)

    def load(self, item_key: str) -> R:
        """Load a single item result, honoring the schema-version stamp.

        Three on-disk shapes are accepted:

        - **Stamped** (current): ``{"_schema_version": "...", "_data": {...}}``
          → version-checked. Mismatch raises ``ValueError``;
          ``partition()`` treats this as a cache miss and re-processes.
        - **Legacy** (pre-stamp): top-level Pydantic JSON
          → loaded via ``model_validate``; surfaces as a "stale" miss
          only if the schema fields no longer match.
        - **Corrupt**: invalid JSON or schema mismatch → raises;
          ``partition()`` catches this and treats as a miss.
        """
        text = self.store.get(self._item_key(item_key))
        if text is None:
            raise FileNotFoundError(f"cache miss for {item_key!r} in namespace {self.namespace!r}")
        raw: Any = json.loads(text)
        if is_dict(raw) and _SCHEMA_VERSION_FIELD in raw and _DATA_FIELD in raw:
            stored_version = raw[_SCHEMA_VERSION_FIELD]
            expected = self._schema_version()
            if stored_version != expected:
                raise ValueError(
                    f"schema_version mismatch for {item_key}: stored={stored_version} expected={expected}"
                )
            return self.schema.model_validate(raw[_DATA_FIELD])
        # Legacy path: older caches stored the Pydantic JSON directly.
        return self.schema.model_validate(raw)

    def partition(self, keys: list[str]) -> tuple[dict[str, R], list[int]]:
        """Split keys into cached results and uncached indices.

        If a cached item fails validation (e.g. schema changed), it is
        treated as a cache miss and will be re-processed. Any in-flight
        markers from a prior crashed run are also reported.
        """
        cached: dict[str, R] = {}
        uncached_indices: list[int] = []
        n_stale = 0
        for index, item_key in enumerate(keys):
            if self.exists(item_key):
                try:
                    cached[item_key] = self.load(item_key)
                except Exception as exc:
                    emit(
                        "item_cache.stale_entry",
                        namespace=self.namespace,
                        item_key=item_key,
                        error_type=type(exc).__name__,
                        error=str(exc)[:200],
                    )
                    uncached_indices.append(index)
                    n_stale += 1
            else:
                uncached_indices.append(index)

        n_inflight = len(self.inflight_keys())
        if cached or n_stale or n_inflight:
            emit(
                "item_cache.partition",
                namespace=self.namespace,
                cached=len(cached),
                uncached=len(uncached_indices),
                stale=n_stale,
                inflight_orphans=n_inflight,
            )

        return cached, uncached_indices

    def collect(self, keys: list[str], *, allow_missing: bool = False) -> list[R]:
        """Load all items by keys.

        Default behavior: raises ``ValueError`` if any key is missing, the
        downstream step probably can't proceed without a complete result set.

        With ``allow_missing=True``: silently skip missing keys and return
        whatever was actually cached. Used by batch steps that tolerate
        partial failure (e.g. LLM calls where a small percentage may
        legitimately fail due to rate limits / transient errors, but the
        successful majority is still worth keeping).
        """
        results: list[R] = []
        missing = 0
        for item_key in keys:
            if not self.exists(item_key):
                if allow_missing:
                    missing += 1
                    continue
                raise ValueError(f"Item '{item_key}' not found in cache '{self.namespace}'")
            results.append(self.load(item_key))
        if missing:
            emit(
                "item_cache.collect.partial",
                namespace=self.namespace,
                kept=len(results),
                missing=missing,
            )
        return results

key

key(*parts: str) -> str

Compute a cache key from the item's identity parts.

Source code in src/adag/item_cache.py

def key(self, *parts: str) -> str:
    """Compute a cache key from the item's identity parts."""
    h = hashlib.sha256()
    for part in parts:
        h.update(part.encode())
    return h.hexdigest()[:16]

exists

exists(item_key: str) -> bool

Check if an item is cached.

Source code in src/adag/item_cache.py

def exists(self, item_key: str) -> bool:
    """Check if an item is cached."""
    return self.store.exists(self._item_key(item_key))

list_keys

list_keys() -> set[str]

Return all cached item keys.

Source code in src/adag/item_cache.py

def list_keys(self) -> set[str]:
    """Return all cached item keys."""
    prefix = f"{self.namespace}/"
    return {
        k[len(prefix) : -len(".json")]
        for k in self.store.iter_keys(prefix)
        if k.endswith(".json")
    }

mark_inflight

mark_inflight(item_key: str) -> None

Record that processing for item_key has started.

Writes a <key>.inflight sentinel. If the worker crashes before save(), this sentinel persists and can be surfaced on next resume to flag work-in-progress items that were lost. Call clear_inflight() after a successful save.

Source code in src/adag/item_cache.py

def mark_inflight(self, item_key: str) -> None:
    """Record that processing for ``item_key`` has started.

    Writes a ``<key>.inflight`` sentinel. If the worker crashes before
    ``save()``, this sentinel persists and can be surfaced on next
    resume to flag work-in-progress items that were lost. Call
    ``clear_inflight()`` after a successful save.
    """
    self.store.set(self._inflight_key(item_key), "")

clear_inflight

clear_inflight(item_key: str) -> None

Remove the in-flight sentinel after a successful save.

Source code in src/adag/item_cache.py

def clear_inflight(self, item_key: str) -> None:
    """Remove the in-flight sentinel after a successful save."""
    self.store.delete(self._inflight_key(item_key))

inflight_keys

inflight_keys() -> set[str]

All currently-marked in-flight item keys (typically from a crashed run).

Source code in src/adag/item_cache.py

def inflight_keys(self) -> set[str]:
    """All currently-marked in-flight item keys (typically from a crashed run)."""
    prefix = f"{self.namespace}/"
    return {
        k[len(prefix) : -len(".inflight")]
        for k in self.store.iter_keys(prefix)
        if k.endswith(".inflight")
    }

save

save(item_key: str, result: R) -> None

Save a single item result with a schema-version stamp.

Stamping the schema version on every write lets a future load() detect when the cached payload was produced under a different schema and refuse to deserialize it. Old, pre-stamp caches still load via the legacy path below for backwards compatibility.

Source code in src/adag/item_cache.py

def save(self, item_key: str, result: R) -> None:
    """Save a single item result with a schema-version stamp.

    Stamping the schema version on every write lets a future
    ``load()`` detect when the cached payload was produced under a
    different schema and refuse to deserialize it. Old, pre-stamp
    caches still load via the legacy path below for backwards
    compatibility.
    """
    wrapper: dict[str, Any] = {
        _SCHEMA_VERSION_FIELD: self._schema_version(),
        _DATA_FIELD: result.model_dump(mode="json"),
    }
    self.store.set(self._item_key(item_key), json.dumps(wrapper, indent=2))
    self.clear_inflight(item_key)

load

load(item_key: str) -> R

Load a single item result, honoring the schema-version stamp.

Three on-disk shapes are accepted:

• Stamped (current): {"_schema_version": "...", "_data": {...}} → version-checked. Mismatch raises ValueError; partition() treats this as a cache miss and re-processes.
• Legacy (pre-stamp): top-level Pydantic JSON → loaded via model_validate; surfaces as a "stale" miss only if the schema fields no longer match.
• Corrupt: invalid JSON or schema mismatch → raises; partition() catches this and treats as a miss.

Source code in src/adag/item_cache.py

def load(self, item_key: str) -> R:
    """Load a single item result, honoring the schema-version stamp.

    Three on-disk shapes are accepted:

    - **Stamped** (current): ``{"_schema_version": "...", "_data": {...}}``
      → version-checked. Mismatch raises ``ValueError``;
      ``partition()`` treats this as a cache miss and re-processes.
    - **Legacy** (pre-stamp): top-level Pydantic JSON
      → loaded via ``model_validate``; surfaces as a "stale" miss
      only if the schema fields no longer match.
    - **Corrupt**: invalid JSON or schema mismatch → raises;
      ``partition()`` catches this and treats as a miss.
    """
    text = self.store.get(self._item_key(item_key))
    if text is None:
        raise FileNotFoundError(f"cache miss for {item_key!r} in namespace {self.namespace!r}")
    raw: Any = json.loads(text)
    if is_dict(raw) and _SCHEMA_VERSION_FIELD in raw and _DATA_FIELD in raw:
        stored_version = raw[_SCHEMA_VERSION_FIELD]
        expected = self._schema_version()
        if stored_version != expected:
            raise ValueError(
                f"schema_version mismatch for {item_key}: stored={stored_version} expected={expected}"
            )
        return self.schema.model_validate(raw[_DATA_FIELD])
    # Legacy path: older caches stored the Pydantic JSON directly.
    return self.schema.model_validate(raw)

partition

partition(
    keys: list[str],
) -> tuple[dict[str, R], list[int]]

Split keys into cached results and uncached indices.

If a cached item fails validation (e.g. schema changed), it is treated as a cache miss and will be re-processed. Any in-flight markers from a prior crashed run are also reported.

Source code in src/adag/item_cache.py

def partition(self, keys: list[str]) -> tuple[dict[str, R], list[int]]:
    """Split keys into cached results and uncached indices.

    If a cached item fails validation (e.g. schema changed), it is
    treated as a cache miss and will be re-processed. Any in-flight
    markers from a prior crashed run are also reported.
    """
    cached: dict[str, R] = {}
    uncached_indices: list[int] = []
    n_stale = 0
    for index, item_key in enumerate(keys):
        if self.exists(item_key):
            try:
                cached[item_key] = self.load(item_key)
            except Exception as exc:
                emit(
                    "item_cache.stale_entry",
                    namespace=self.namespace,
                    item_key=item_key,
                    error_type=type(exc).__name__,
                    error=str(exc)[:200],
                )
                uncached_indices.append(index)
                n_stale += 1
        else:
            uncached_indices.append(index)

    n_inflight = len(self.inflight_keys())
    if cached or n_stale or n_inflight:
        emit(
            "item_cache.partition",
            namespace=self.namespace,
            cached=len(cached),
            uncached=len(uncached_indices),
            stale=n_stale,
            inflight_orphans=n_inflight,
        )

    return cached, uncached_indices

collect

collect(
    keys: list[str], *, allow_missing: bool = False
) -> list[R]

Load all items by keys.

Default behavior: raises ValueError if any key is missing, the downstream step probably can't proceed without a complete result set.

With allow_missing=True: silently skip missing keys and return whatever was actually cached. Used by batch steps that tolerate partial failure (e.g. LLM calls where a small percentage may legitimately fail due to rate limits / transient errors, but the successful majority is still worth keeping).

Source code in src/adag/item_cache.py

def collect(self, keys: list[str], *, allow_missing: bool = False) -> list[R]:
    """Load all items by keys.

    Default behavior: raises ``ValueError`` if any key is missing, the
    downstream step probably can't proceed without a complete result set.

    With ``allow_missing=True``: silently skip missing keys and return
    whatever was actually cached. Used by batch steps that tolerate
    partial failure (e.g. LLM calls where a small percentage may
    legitimately fail due to rate limits / transient errors, but the
    successful majority is still worth keeping).
    """
    results: list[R] = []
    missing = 0
    for item_key in keys:
        if not self.exists(item_key):
            if allow_missing:
                missing += 1
                continue
            raise ValueError(f"Item '{item_key}' not found in cache '{self.namespace}'")
        results.append(self.load(item_key))
    if missing:
        emit(
            "item_cache.collect.partial",
            namespace=self.namespace,
            kept=len(results),
            missing=missing,
        )
    return results

ComponentKind

Bases: StrEnum

The kind of one input to a cache key.

Source code in src/adag/key.py

class ComponentKind(StrEnum):
    """The kind of one input to a cache key."""

    IDENTITY = "identity"
    """The step's class name or ``kind``; for a node, the step's own key."""
    CODE = "code"
    """A digest of the step class's source code."""
    PINNED = "pinned"
    """``Step.cache_key`` is set. It replaces identity, code and fields."""
    FIELD = "field"
    """A step field, encoded by value."""
    FILE = "file"
    """A ``FileSource`` field, hashed by the file's content."""
    NESTED_STEP = "nested_step"
    """A field that holds another step, hashed by that step's key."""
    UPSTREAM = "upstream"
    """An input wired to another node, hashed by that node's key."""
    PARAM = "param"
    """An input wired to an ``InputParam``, hashed by the run's value or the default."""
    DEFERRED = "deferred"
    """An input wired to another source, such as a ``FileSource``."""
    RAW = "raw"
    """An input passed as a literal value."""
    SKIPPED = "skipped"
    """Not part of the key: a ``Config`` field, ``node_label``, or a value that cannot be
    hashed when ``strict_content_hash = False``."""

IDENTITY class-attribute instance-attribute

IDENTITY = 'identity'

The step's class name or kind; for a node, the step's own key.

CODE class-attribute instance-attribute

CODE = 'code'

A digest of the step class's source code.

PINNED class-attribute instance-attribute

PINNED = 'pinned'

Step.cache_key is set. It replaces identity, code and fields.

FIELD class-attribute instance-attribute

FIELD = 'field'

A step field, encoded by value.

FILE class-attribute instance-attribute

FILE = 'file'

A FileSource field, hashed by the file's content.

NESTED_STEP class-attribute instance-attribute

NESTED_STEP = 'nested_step'

A field that holds another step, hashed by that step's key.

UPSTREAM class-attribute instance-attribute

UPSTREAM = 'upstream'

An input wired to another node, hashed by that node's key.

PARAM class-attribute instance-attribute

PARAM = 'param'

An input wired to an InputParam, hashed by the run's value or the default.

DEFERRED class-attribute instance-attribute

DEFERRED = 'deferred'

An input wired to another source, such as a FileSource.

RAW class-attribute instance-attribute

RAW = 'raw'

An input passed as a literal value.

SKIPPED class-attribute instance-attribute

SKIPPED = 'skipped'

Not part of the key: a Config field, node_label, or a value that cannot be hashed when strict_content_hash = False.

KeyComponent dataclass

One input to a cache key.

digest is the first 12 hex characters of sha256 over the bytes this component adds to the key. Two components with the same (kind, name) and the same digest added the same bytes. It is empty for a SKIPPED component and for one that blocked the key.

Source code in src/adag/key.py

@dataclass(frozen=True)
class KeyComponent:
    """One input to a cache key.

    ``digest`` is the first 12 hex characters of sha256 over the bytes this component adds to the
    key. Two components with the same ``(kind, name)`` and the same digest added the same bytes.
    It is empty for a ``SKIPPED`` component and for one that blocked the key.
    """

    kind: ComponentKind
    name: str
    value: str
    digest: str

    @staticmethod
    def build(kind: ComponentKind, name: str, value: Any, chunk: bytes | None) -> KeyComponent:
        """Build a component from one ``_key_chunks`` tuple."""
        digest = hashlib.sha256(chunk).hexdigest()[:12] if chunk else ""
        return KeyComponent(kind=kind, name=name, value=_value_text(value), digest=digest)

build staticmethod

build(
    kind: ComponentKind,
    name: str,
    value: Any,
    chunk: bytes | None,
) -> KeyComponent

Build a component from one _key_chunks tuple.

Source code in src/adag/key.py

@staticmethod
def build(kind: ComponentKind, name: str, value: Any, chunk: bytes | None) -> KeyComponent:
    """Build a component from one ``_key_chunks`` tuple."""
    digest = hashlib.sha256(chunk).hexdigest()[:12] if chunk else ""
    return KeyComponent(kind=kind, name=name, value=_value_text(value), digest=digest)

KeyExplanation dataclass

Every component of one node's cache key, in the order adag hashes them.

key is None when a component blocked it (a missing file, an upstream with no key); blocked_by then names that component. An UPSTREAM component shows the upstream step and its key: explain that node next to follow a change up the graph.

Source code in src/adag/key.py

@dataclass(frozen=True)
class KeyExplanation:
    """Every component of one node's cache key, in the order adag hashes them.

    ``key`` is ``None`` when a component blocked it (a missing file, an upstream with no key);
    ``blocked_by`` then names that component. An ``UPSTREAM`` component shows the upstream step
    and its key: explain that node next to follow a change up the graph.
    """

    step: str
    label: str | None
    key: str | None
    blocked_by: str | None
    key_format: int
    components: tuple[KeyComponent, ...]

    def redacted(self) -> KeyExplanation:
        """A copy with the value of every ``SKIPPED`` component removed.

        A skipped component is not in the key, and its value can be a client, a path or a
        secret (a ``Config`` field). The key record is written to disk or to a shared store, so
        it keeps only the name.
        """
        return replace(
            self,
            components=tuple(
                replace(c, value="") if c.kind is ComponentKind.SKIPPED else c
                for c in self.components
            ),
        )

    def to_dict(self) -> dict[str, Any]:
        """The JSON form, as written to the key record."""
        data = asdict(self)
        data["components"] = [{**asdict(c), "kind": c.kind.value} for c in self.components]
        return data

    @staticmethod
    def from_dict(data: dict[str, Any]) -> KeyExplanation:
        """Rebuild an explanation from ``to_dict`` output."""
        return KeyExplanation(
            step=data["step"],
            label=data["label"],
            key=data["key"],
            blocked_by=data["blocked_by"],
            key_format=data["key_format"],
            components=tuple(
                KeyComponent(
                    kind=ComponentKind(c["kind"]),
                    name=c["name"],
                    value=c["value"],
                    digest=c["digest"],
                )
                for c in data["components"]
            ),
        )

    def __str__(self) -> str:
        title = f"{self.step}" + (f" ({self.label})" if self.label else "")
        status = self.key if self.key is not None else f"blocked by {self.blocked_by}"
        rows = [(c.kind.value, c.name or "-", c.value, c.digest or "-") for c in self.components]
        widths = [max([len(r[i]) for r in rows] + [1]) for i in range(3)]
        lines = [f"{title}  key={status}  format={self.key_format}"]
        for kind, name, value, digest in rows:
            lines.append(
                f"  {kind:<{widths[0]}}  {name:<{widths[1]}}  {value:<{widths[2]}}  {digest}"
            )
        return "\n".join(lines)

redacted

redacted() -> KeyExplanation

A copy with the value of every SKIPPED component removed.

A skipped component is not in the key, and its value can be a client, a path or a secret (a Config field). The key record is written to disk or to a shared store, so it keeps only the name.

Source code in src/adag/key.py

def redacted(self) -> KeyExplanation:
    """A copy with the value of every ``SKIPPED`` component removed.

    A skipped component is not in the key, and its value can be a client, a path or a
    secret (a ``Config`` field). The key record is written to disk or to a shared store, so
    it keeps only the name.
    """
    return replace(
        self,
        components=tuple(
            replace(c, value="") if c.kind is ComponentKind.SKIPPED else c
            for c in self.components
        ),
    )

to_dict

to_dict() -> dict[str, Any]

The JSON form, as written to the key record.

Source code in src/adag/key.py

def to_dict(self) -> dict[str, Any]:
    """The JSON form, as written to the key record."""
    data = asdict(self)
    data["components"] = [{**asdict(c), "kind": c.kind.value} for c in self.components]
    return data

from_dict staticmethod

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

Rebuild an explanation from to_dict output.

Source code in src/adag/key.py

@staticmethod
def from_dict(data: dict[str, Any]) -> KeyExplanation:
    """Rebuild an explanation from ``to_dict`` output."""
    return KeyExplanation(
        step=data["step"],
        label=data["label"],
        key=data["key"],
        blocked_by=data["blocked_by"],
        key_format=data["key_format"],
        components=tuple(
            KeyComponent(
                kind=ComponentKind(c["kind"]),
                name=c["name"],
                value=c["value"],
                digest=c["digest"],
            )
            for c in data["components"]
        ),
    )

LoggingObserver

Bases: NullObserver

The default observer (installed by Pipeline.execute): logs each step's lifecycle as a structlog event.

This is the single place per-step logging happens. resolve emits the events and this turns them into log lines. To make one event louder or quieter, change its level here (one place). For full silence, pass NullObserver.

Source code in src/adag/observers.py

class LoggingObserver(NullObserver):
    """The default observer (installed by ``Pipeline.execute``): logs each step's
    lifecycle as a structlog event.

    This is the single place per-step logging happens. ``resolve`` emits the
    events and this turns them into log lines. To make one event louder or quieter,
    change its level here (one place). For full silence, pass ``NullObserver``.
    """

    def on_step_start(self, step: str) -> None:
        logger.info("step.start", step=step)

    def on_step_cached(self, step: str) -> None:
        logger.info("step.cache_hit", step=step)

    def on_step_finished(self, step: str, seconds: float) -> None:
        logger.info("step.finished", step=step, seconds=round(seconds, 4))

    def on_event(self, event: str, fields: dict[str, Any]) -> None:
        logger.info(event, **fields)

RunTreeObserver

Bases: NullObserver

Builds an indented tree of a single pipeline run, annotated with timing and cache-hit markers.

Render it with render_tree() after execute(). Use one observer per execute(): it is not reusable across runs. It is safe under threads: per-thread stacks keep each worker's bracketing consistent, and roots merge under a lock. Under parallel=True the tree is flat, because level-scheduled execution has no recursion to nest.

Source code in src/adag/observers.py

class RunTreeObserver(NullObserver):
    """Builds an indented tree of a single pipeline run, annotated with timing and cache-hit markers.

    Render it with ``render_tree()`` after ``execute()``. Use one observer per ``execute()``: it
    is not reusable across runs. It is safe under threads: per-thread stacks keep each worker's
    bracketing consistent, and roots merge under a lock. Under ``parallel=True`` the tree is flat,
    because level-scheduled execution has no recursion to nest.
    """

    def __init__(self) -> None:
        self._roots: list[RunNode] = []
        self._roots_lock = threading.Lock()
        self._state = _ThreadRunState()

    def on_fan_out_branch(self, key: str) -> None:
        # The next step to start is the root of this fan-out branch; label it by key.
        self._state.pending_label = key

    def on_step_start(self, step: str) -> None:
        label = self._state.pending_label
        name = step if label is None else f"{step} [{label}]"
        self._state.pending_label = None
        node = RunNode(name=name)
        stack = self._state.stack
        if stack:
            stack[-1].children.append(node)
        else:
            with self._roots_lock:
                self._roots.append(node)
        stack.append(node)

    def on_step_cached(self, step: str) -> None:
        node = self._state.stack.pop()
        node.cached = True

    def on_step_finished(self, step: str, seconds: float) -> None:
        node = self._state.stack.pop()
        node.seconds = seconds

    def render_tree(self) -> str:
        """Render the collected run as an indented tree string."""
        lines: list[str] = []
        for root in self._roots:
            self._render(root, prefix="", connector="", lines=lines)
        return "\n".join(lines)

    def _render(self, node: RunNode, prefix: str, connector: str, lines: list[str]) -> None:
        lines.append(f"{prefix}{connector}{node.name}  {self._marker(node)}")
        child_prefix = (
            prefix if not connector else prefix + ("   " if connector == "└─ " else "│  ")
        )
        for index, child in enumerate(node.children):
            is_last = index == len(node.children) - 1
            self._render(child, child_prefix, "└─ " if is_last else "├─ ", lines)

    def _marker(self, node: RunNode) -> str:
        if node.cached:
            return "⟳ cached"
        if node.seconds is None:
            return "● (unfinished)"
        if node.seconds < 1.0:
            return f"● {node.seconds * 1000:.0f}ms"
        return f"● {node.seconds:.1f}s"

render_tree

render_tree() -> str

Render the collected run as an indented tree string.

Source code in src/adag/observers.py

def render_tree(self) -> str:
    """Render the collected run as an indented tree string."""
    lines: list[str] = []
    for root in self._roots:
        self._render(root, prefix="", connector="", lines=lines)
    return "\n".join(lines)

CyclicPipelineError

Bases: ValueError

Raised at construction when the Deferred graph contains a cycle (a step transitively depends on itself).

Source code in src/adag/pipeline.py

class CyclicPipelineError(ValueError):
    """Raised at construction when the Deferred graph contains a cycle (a step transitively depends on itself)."""

Pipeline

Bases: Generic[Out]

Wraps a Deferred graph. Discovers steps and parameters, provides execute().

tokenizer = Tokenizer(max_length=256) text = InputParam("text") tokens = tokenizer(text=text) pipeline = Pipeline(tokens, cache_dir=Path("cache")) result = pipeline.execute(text="hello world")

With neither cache_dir nor store (the default), nothing is cached and every execute() runs the full graph. Pass store to cache in a CacheStore instead of files under cache_dir.

Source code in src/adag/pipeline.py

class Pipeline(Generic[Out]):
    """Wraps a Deferred graph. Discovers steps and parameters, provides execute().

    tokenizer = Tokenizer(max_length=256)
    text = InputParam("text")
    tokens = tokenizer(text=text)
    pipeline = Pipeline(tokens, cache_dir=Path("cache"))
    result = pipeline.execute(text="hello world")

    With neither ``cache_dir`` nor ``store`` (the default), nothing is cached and every
    ``execute()`` runs the full graph. Pass ``store`` to cache in a ``CacheStore`` instead of
    files under ``cache_dir``.
    """

    def __init__(
        self,
        output: StepResult[Out],
        cache_dir: Path | None = None,
        on_progress: Callable[[Progress], None] | None = None,
        *,
        kind: str | None = None,
        extra_targets: list[StepResult[Any]] | None = None,
        store: CacheStore | None = None,
        metadata: JsonValue = None,
    ) -> None:
        """Wrap a Deferred graph.

        Args:
            output: the canonical terminal node of the DAG. ``execute()``
                without an argument resolves this.
            cache_dir: directory for step results. With neither ``cache_dir`` nor ``store``,
                nothing is cached.
            store: a ``CacheStore`` for step results and ``BatchStep`` item caches, used
                instead of files under ``cache_dir``.
            on_progress: optional progress callback (passed to ``ResolveContext``).
            kind: a tag for the pipeline, for example ``"report/v1"``. ``save()`` writes it to
                ``pipeline.json``.
            metadata: caller data saved with the pipeline: any JSON value (a string, a number, a
                list, a dict). ``save()`` writes it to ``pipeline.json`` and ``from_directory()``
                reads it back into ``pipeline.metadata``. adag never reads it otherwise.
            extra_targets: additional terminal or intermediate nodes whose
                sub-DAGs should be discovered too. Use this when the
                pipeline has multiple "ending points" so visualization and
                ``resolve()`` can reach every step in the graph (not just
                the ones reachable from ``output``). Use
                ``Pipeline.from_targets([...])`` for a slightly nicer call
                site when you have no canonical terminal.
        """
        self.output = output
        self.cache_dir = cache_dir
        self.store = store
        self.on_progress = on_progress
        self.kind = kind
        if metadata is not None:
            try:
                json.dumps(metadata)
            except (TypeError, ValueError) as exc:
                raise TypeError(f"Pipeline metadata must be a JSON value: {exc}") from exc
        self.metadata: JsonValue = metadata

        self._steps: list[Step[Any]] = []
        self._inputs: list[InputParam[Any]] = []
        self._targets: list[StepResult[Any]] = [output]
        self._discover(output)
        for extra in extra_targets or ():
            if extra is not output and extra not in self._targets:
                self._targets.append(extra)
                self._discover(extra)

    @classmethod
    def from_targets(
        cls,
        targets: list[StepResult[Any]],
        cache_dir: Path | None = None,
        on_progress: Callable[[Progress], None] | None = None,
        *,
        kind: str | None = None,
        store: CacheStore | None = None,
        metadata: JsonValue = None,
    ) -> Pipeline[Any]:
        """Build a Pipeline that covers multiple terminal nodes.

        Real DAGs often have more than one ending point (e.g. the main
        result plus a separate audit / dedup artifact). Pass them all
        here and the Pipeline discovers the union of their sub-graphs.
        The first target is treated as the canonical ``output`` so a
        zero-arg ``execute()`` still has a single thing to return.
        Downstream callers use ``resolve(target)`` to pick any of them.
        """
        if not targets:
            raise ValueError("Pipeline.from_targets requires at least one target.")
        head, *rest = targets
        return cls(
            output=head,
            cache_dir=cache_dir,
            on_progress=on_progress,
            kind=kind,
            extra_targets=rest,
            store=store,
            metadata=metadata,
        )

    @property
    def targets(self) -> list[StepResult[Any]]:
        """All terminal / target nodes the pipeline was built to resolve."""
        return list(self._targets)

    def _discover(self, deferred: Deferred[Any], visiting: set[int] | None = None) -> None:
        """Walk the Deferred graph backward to collect all steps and parameters.

        ``visiting`` is the current DFS path (StepResult ids). Re-entering a node already
        on the path means a cycle, we raise CyclicPipelineError instead of letting the
        recursion blow up as an opaque RecursionError deep in resolve(). Steps are appended
        after their sources, so ``self._steps`` ends up in dependency (topological) order.
        """
        visiting = visiting if visiting is not None else set()
        if isinstance(deferred, InputParam) and deferred not in self._inputs:
            self._inputs.append(deferred)
        elif isinstance(deferred, StepResult):
            if id(deferred) in visiting:
                label = type(deferred.step).label or type(deferred.step).__name__
                raise CyclicPipelineError(f"Pipeline graph contains a cycle through step {label!r}")
            visiting.add(id(deferred))
            for source in deferred.sources.values():
                if isinstance(source, Deferred):
                    self._discover(source, visiting)
            visiting.discard(id(deferred))
            if deferred.step not in self._steps:
                self._steps.append(deferred.step)

    @property
    def steps(self) -> list[Step[Any]]:
        """All steps in dependency (topological) order.

        Every step appears after the steps it depends on. The post-order discovery walk in
        ``_discover`` guarantees it.
        """
        return list(self._steps)

    def topological_order(self) -> list[Step[Any]]:
        """Steps in a guaranteed topological order (each step after its dependencies).

        A named contract (delegates to ``steps``). Consumers that depend on the ordering
        (deterministic execution, left-to-right layered layout, level grouping) rely on this
        rather than on the ``steps`` discovery detail.
        """
        return self.steps

    def topological_levels(self) -> list[list[Step[Any]]]:
        """Steps grouped into dependency levels: every step lands one level past its deepest
        dependency. Steps within a level are mutually independent, so a concurrent executor
        can run a whole level in parallel; level i+1 starts once level i is resolved.

        Pure analysis: no execution change. The foundation for opt-in parallel execution
        (run each level's steps together) and for correct left-to-right layered layout.
        """
        # step -> upstream steps, walked from the targets by StepResult id (no cached map).
        upstreams: defaultdict[Step[Any], set[Step[Any]]] = defaultdict(set)
        seen: set[int] = set()
        stack: list[StepResult[Any]] = list(self._targets)
        while stack:
            node = stack.pop()
            if id(node) in seen:
                continue
            seen.add(id(node))
            for source in node.sources.values():
                if isinstance(source, StepResult):
                    upstreams[node.step].add(source.step)
                    stack.append(source)

        # self._steps is already topological, so each step's upstreams have a level by now.
        level_of: dict[Step[Any], int] = {}
        levels: list[list[Step[Any]]] = []
        for step in self._steps:
            dependencies = upstreams.get(step, set())
            level = (
                0
                if not dependencies
                else 1 + max(level_of[dependency] for dependency in dependencies)
            )
            level_of[step] = level
            if level == len(levels):
                levels.append([])
            levels[level].append(step)
        return levels

    def _stepresult_levels(self) -> list[list[StepResult[Any]]]:
        """The graph's StepResult nodes grouped into dependency levels (Kahn over node ids)."""
        return _node_levels(self._targets)

    def _resolve_levels_concurrently(
        self, context: ResolveContext, max_workers: int | None
    ) -> None:
        """Resolve the whole graph level by level, each level's nodes in a thread pool.

        Steps within a level are independent, so they run concurrently; the next level
        starts only once this one is fully resolved. ``resolve`` memoises by ``id`` and
        every node's upstreams are resolved in earlier levels, so each node finds its inputs
        already in ``context._resolved`` and just runs its own step, no recursion, no step
        run twice.

        Each StepResult is resolved exactly once (in its level), with upstream keys written
        in earlier levels and only read here. ``context._lock`` (see ResolveContext) guards
        every ``_resolved`` read/write, so concurrent nodes are safe regardless of the GIL. It
        does not rely on CPython dict-op atomicity, which the free-threaded build removes.

        Each node runs inside a copied context (``copy_context().run``), so ambient ContextVars
        (the usage observer, ``current_observer``) propagate into the level workers, which a raw
        ThreadPoolExecutor submit would not inherit.
        """
        for level in self._stepresult_levels():
            if len(level) == 1:
                level[0].resolve(context)
                continue
            with ThreadPoolExecutor(max_workers=max_workers or len(level)) as pool:
                futures = [pool.submit(copy_context().run, node.resolve, context) for node in level]
                for future in as_completed(futures):
                    future.result()

    def explain_keys(self, params: dict[str, Any] | None = None) -> tuple[KeyExplanation, ...]:
        """Explain the cache key of every node, upstream nodes first.

        Args:
            params: the run's params, as passed to ``execute``.
        """
        return tuple(
            node.explain_key(params) for level in self._stepresult_levels() for node in level
        )

    def validate(self) -> None:
        """Validate the graph before ``execute()``: every step's required run() input is bound.

        Cycles are already rejected at construction (see ``_discover``). This adds the
        missing-input check: a required parameter of a step's ``run()`` that is neither a
        wired source nor one of the step's dataclass fields would otherwise fail late at
        ``execute()`` as a bare ``TypeError: run() missing 1 required argument``. Opt-in
        (not run at construction) so a consumer building a graph from dynamic/untrusted
        input can validate it up front and reject with a 4xx instead of crashing
        mid-resolution. Raises ``UnboundInputError`` naming the step and the unbound params.

        Walks the target StepResults directly (visited by id, the same node identity
        ``resolve()`` uses) rather than caching a derived step->sources map.
        """
        seen: set[int] = set()
        stack: list[StepResult[Any]] = list(self._targets)
        while stack:
            node = stack.pop()
            if id(node) in seen:
                continue
            seen.add(id(node))
            self._check_inputs_bound(node)
            stack.extend(
                source for source in node.sources.values() if isinstance(source, StepResult)
            )

    def _check_inputs_bound(self, node: StepResult[Any]) -> None:
        """Raise UnboundInputError if ``node.step`` has a required ``run()`` parameter that is
        neither a wired source nor one of its dataclass fields, both feed ``run()`` (see
        ``StepResult._merge_step_defaults``), so either satisfies the parameter.
        """
        field_names = {field.name for field in dataclasses.fields(node.step)}
        available = set(node.sources.keys()) | field_names
        try:
            signature = inspect.signature(type(node.step).run)
        except (ValueError, TypeError):
            return
        required: set[str] = set()
        for name, parameter in signature.parameters.items():
            if name == "self" or parameter.kind in (
                inspect.Parameter.VAR_KEYWORD,
                inspect.Parameter.VAR_POSITIONAL,
            ):
                continue
            if parameter.default is inspect.Parameter.empty:
                required.add(name)
        missing = required - available
        if missing:
            label = type(node.step).label or type(node.step).__name__
            raise UnboundInputError(
                f"Step {label!r} has unbound required input(s): {sorted(missing)}"
            )

    @property
    def input_names(self) -> set[str]:
        """Names of every InputParam this pipeline expects from execute(**kwargs).

        Public alternative to reaching into ``_inputs`` from outside the package. Callers that
        branch on "does this pipeline accept a `threshold` kwarg" should use this, so the
        private list shape stays free to evolve.
        """
        return {p.name for p in self._inputs}

    @property
    def inputs(self) -> list[InputParam[Any]]:
        """Every InputParam this pipeline expects, in discovery order.

        Public accessor for callers (e.g. the CLI) that need each param's name,
        type, and default (not just the names from ``input_names``) to build
        help text or coerce string arguments.
        """
        return list(self._inputs)

    def step_name(self, step: Step[Any]) -> str:
        """Directory name for a step: {index}_{name}, falling back to the lowercased class name."""
        index = self._steps.index(step)
        width = max(2, math.floor(math.log10(max(len(self._steps), 1))) + 1)
        name = getattr(step, "name", None) or type(step).__name__.lower()
        return f"{index:0{width}d}_{name}"

    def item_cache(self, namespace: str, schema: type[R]) -> ItemCache[R]:
        """Create an ItemCache in this pipeline's ``store``, or under its cache directory."""
        if self.store is not None:
            return ItemCache(self.cache_dir or Path(), namespace, schema, store=self.store)
        if self.cache_dir is None:
            raise ValueError("Cannot create ItemCache without a cache directory or store.")
        return ItemCache(self.cache_dir, namespace, schema)

    def execute(
        self,
        *,
        observer: PipelineObserver | None = None,
        executor: Executor | None = None,
        dispatcher: BranchDispatcher | None = None,
        parallel: bool = False,
        max_workers: int | None = None,
        **params: Any,
    ) -> Out:
        """Run the graph and return the result of ``output``.

        Args:
            observer: receives the run's hooks. Default: ``LoggingObserver``. Pass ``NullObserver``
                for silence.
            executor: where each step runs. Default: ``LocalExecutor`` (this process).
            dispatcher: where fan-out branches run. Default: ``InlineDispatcher`` (this process).
                Pass a ``SubprocessDispatcher`` or a ``RemoteDispatcher`` to run them elsewhere.
            parallel: run the independent steps of each level on a thread pool. Use it for steps
                that block on I/O. ``RunTreeObserver`` then shows a flat tree.
            max_workers: thread pool size for ``parallel``. Default: one thread per step of the level.
            **params: one value per ``InputParam``.

        The five keyword names above are reserved: do not use them as ``InputParam`` names.

        Raises:
            ValueError: a required ``InputParam`` has no value.
        """
        missing = [p.name for p in self._inputs if p.name not in params and not p.has_default]
        if missing:
            raise ValueError(f"Missing parameters: {missing}")

        context = ResolveContext(
            cache_dir=self.cache_dir,
            store=self.store,
            params=params,
            on_progress=self.on_progress,
            observer=observer or LoggingObserver(),
            executor=executor or LocalExecutor(),
            dispatcher=dispatcher or InlineDispatcher(),
        )
        with observing(context.observer):
            if parallel:
                self._resolve_levels_concurrently(context, max_workers)
            return self.output.resolve(context)

    async def aexecute(
        self,
        *,
        observer: PipelineObserver | None = None,
        async_executor: AsyncExecutor | None = None,
        **params: Any,
    ) -> Out:
        """Run the graph on the running event loop and return the result of ``output``.

        ``AsyncStep`` bodies are awaited on the loop. Sync step bodies also run on the loop and block
        it while they run. One step body runs at a time (``aresolve_all`` with ``max_concurrency=1``),
        so a shared resource such as one database session is never used by two steps at once.
        Cancelling the task starts no further step.

        ``observer`` and ``async_executor`` are reserved: do not use them as ``InputParam`` names.
        There is no ``dispatcher``: fan-out branches run in this process.
        """
        missing = [p.name for p in self._inputs if p.name not in params and not p.has_default]
        if missing:
            raise ValueError(f"Missing parameters: {missing}")

        context = ResolveContext(
            cache_dir=self.cache_dir,
            store=self.store,
            params=params,
            on_progress=self.on_progress,
            observer=observer or LoggingObserver(),
            async_executor=async_executor or LocalAsyncExecutor(),
        )
        return (await aresolve_all([self.output], context, max_concurrency=1))[0]

    def resolve(
        self,
        target: StepResult[R],
        *,
        observer: PipelineObserver | None = None,
        executor: Executor | None = None,
        dispatcher: BranchDispatcher | None = None,
        **params: Any,
    ) -> R:
        """Resolve one node of this pipeline and return its result, from the cache when it is there.

        Use it to read an intermediate result after ``execute()``. Do not call ``step.run()`` for
        that: it skips the cache and can return a value that differs from the one downstream steps
        used.

        Args:
            target: a node of this pipeline (the object you wired).
            **params: the ``InputParam`` values, as for ``execute``.

        Raises:
            ValueError: ``target`` is not part of this pipeline, or a required param is missing.
        """
        # `_steps` already contains every step reachable from any registered
        # target after _discover ran, so this check covers multi-target
        # pipelines built via from_targets() too.
        if target.step not in self._steps:
            raise ValueError(
                f"target step {type(target.step).__name__} is not part of this pipeline's DAG"
            )
        # Missing-param check is the same as execute() but scoped to the
        # InputParams reachable from `target`. We don't bother pruning:
        # an extra missing-param check is cheap and avoids surprises.
        missing = [p.name for p in self._inputs if p.name not in params and not p.has_default]
        if missing:
            raise ValueError(f"Missing parameters: {missing}")
        context = ResolveContext(
            cache_dir=self.cache_dir,
            store=self.store,
            params=params,
            on_progress=self.on_progress,
            observer=observer or LoggingObserver(),
            executor=executor or LocalExecutor(),
            dispatcher=dispatcher or InlineDispatcher(),
        )
        with observing(context.observer):
            return target.resolve(context)

    def save(self, path: str | Path) -> None:
        """Save all steps and the pipeline graph to a directory."""
        path = str(path)
        os.makedirs(path, exist_ok=True)

        instance_to_name: dict[int, str] = {}
        step_entries: list[str] = []

        for step in self._steps:
            name = self.step_name(step)
            instance_to_name[id(step)] = name
            step_entries.append(name)

            step_dir = os.path.join(path, name)
            # step.json is owned by Step.save_artifacts (the base writes it via
            # _serialize_step_field; SerializableStep overrides to keep the
            # legacy model_dump format that load_spec() reads). One write per
            # step, no double-write, no Pipeline.save guessing what the spec
            # should look like.
            step.save_artifacts(step_dir)

        edges: list[dict[str, str]] = []
        self._collect_edges(self.output, instance_to_name, edges)

        parameters: dict[str, dict[str, Any]] = {}
        for inp in self._inputs:
            if inp.value_type is not None:
                type_name = inp.value_type.__name__
            else:
                type_name = type(inp.default).__name__ if inp.has_default else "str"
            entry: dict[str, Any] = {"type": type_name}
            if inp.has_default:
                # Reuse the same typed-value coercion as step fields so a default
                # that's a Path / UUID / Enum / BaseModel round-trips cleanly.
                _, default_json = _serialize_step_field(inp.default)
                entry["default"] = default_json if default_json is not None else inp.default
            parameters[inp.name] = entry

        manifest: dict[str, Any] = {"steps": step_entries, "edges": edges, "parameters": parameters}
        if self.kind is not None:
            manifest["kind"] = self.kind
        if self.metadata is not None:
            manifest["metadata"] = self.metadata
        manifest["output"] = instance_to_name[id(self.output.step)]
        with atomic_writer(Path(path).joinpath("pipeline.json")) as f:
            json.dump(manifest, f, indent=2)
            f.write("\n")

        logger.info("Pipeline saved", path=path, steps=len(step_entries))

    @classmethod
    def from_directory(
        cls,
        path: str | Path,
        step_registry: dict[str, type[Step[Any]]],
        *,
        cache_dir: Path | None = None,
    ) -> Pipeline:
        """Reconstruct a saved Pipeline from disk.

        Reads pipeline.json, loads each step via its class in `step_registry` (keyed by kind),
        replays the edge list to rebuild the DAG, and wraps the terminal StepResult.

        cache_dir defaults to None: loaded pipelines do **not** write step results
        back to disk. Pass cache_dir=Path(path) (or another writable location) when
        you want the cross-run cache for training or notebook iteration.
        """
        base = Path(path)
        with open(base.joinpath("pipeline.json")) as f:
            manifest: dict[str, Any] = json.load(f)

        kind: str | None = manifest.get("kind")
        step_entries: list[str] = list(manifest.get("steps", []))
        edges: list[dict[str, str]] = list(manifest.get("edges", []))
        parameters: dict[str, dict[str, Any]] = dict(manifest.get("parameters", {}))
        output_name: str | None = manifest.get("output")

        inputs: dict[str, InputParam[Any]] = {}
        for param_name, entry in parameters.items():
            if "default" in entry:
                inputs[param_name] = InputParam.with_default(param_name, entry["default"])
            else:
                inputs[param_name] = InputParam(name=param_name)

        loaded_steps: dict[str, Step[Any]] = {}
        for step_name in step_entries:
            step_dir = base.joinpath(step_name)
            with open(step_dir.joinpath("step.json")) as f:
                step_kind: str = json.load(f)["kind"]
            step_cls = step_registry.get(step_kind)
            if step_cls is None:
                known = ", ".join(sorted(step_registry.keys()))
                raise ValueError(f"Unknown step kind {step_kind!r} at {step_dir}. Known: [{known}]")
            loaded_steps[step_name] = step_cls.from_directory(str(step_dir))  # type: ignore[attr-defined]

        results: dict[str, StepResult[Any]] = {}
        for step_name in step_entries:
            step = loaded_steps[step_name]
            inputs_for_step: dict[str, Any] = {}
            for edge in edges:
                target = edge.get("target", "")
                if not target.startswith(f"{step_name}."):
                    continue
                input_name = target.split(".", 1)[1]
                source = edge.get("source", "")
                if source.startswith("parameter:"):
                    param_name = source.split(":", 1)[1]
                    inputs_for_step[input_name] = inputs[param_name]
                else:
                    inputs_for_step[input_name] = results[source]
            results[step_name] = step(**inputs_for_step)

        terminal_name = output_name or step_entries[-1]
        output = results[terminal_name]
        return cls(output, cache_dir=cache_dir, kind=kind, metadata=manifest.get("metadata"))

    def _collect_edges(
        self, result: StepResult[Any], instance_to_name: dict[int, str], edges: list[dict[str, str]]
    ) -> None:
        """Recursively collect edges from the StepResult graph."""
        target = instance_to_name.get(id(result.step), "?")
        for input_name, source in result.sources.items():
            if isinstance(source, StepResult):
                source_name = instance_to_name.get(id(source.step), "?")
                edge = {"source": source_name, "target": f"{target}.{input_name}"}
                if edge not in edges:
                    edges.append(edge)
                self._collect_edges(source, instance_to_name, edges)
            elif isinstance(source, InputParam):
                edge = {"source": f"parameter:{source.name}", "target": f"{target}.{input_name}"}
                if edge not in edges:
                    edges.append(edge)

    def status(self) -> None:
        """Log status of all steps."""
        for step in self._steps:
            name = self.step_name(step)
            where = self.store if self.store is not None else self.cache_dir
            if step.result_model() is not None and where is not None:
                step_result = self._find_result(step)
                if step_result and step_result.is_cached(where):
                    logger.info("step.cached", step=name)
                else:
                    logger.info("step.pending", step=name)
            else:
                logger.info("step.ready", step=name)

    def _find_result(self, target_step: Step[Any]) -> StepResult[Any] | None:
        """Find the StepResult for a given step in the graph."""
        return self._walk_find(self.output, target_step)

    def _walk_find(self, deferred: Deferred[Any], target_step: Step[Any]) -> StepResult[Any] | None:
        if isinstance(deferred, StepResult):
            if deferred.step is target_step:
                return deferred
            for source in deferred.sources.values():
                if isinstance(source, Deferred):
                    found = self._walk_find(source, target_step)
                    if found is not None:
                        return found
        return None

    def __repr__(self) -> str:
        steps = [type(s).__name__ for s in self._steps]
        inputs = [p.name for p in self._inputs]
        return f"Pipeline(steps={steps}, inputs={inputs}, cache_dir={self.cache_dir})"

targets property

targets: list[StepResult[Any]]

All terminal / target nodes the pipeline was built to resolve.

steps property

steps: list[Step[Any]]

All steps in dependency (topological) order.

Every step appears after the steps it depends on. The post-order discovery walk in _discover guarantees it.

input_names property

input_names: set[str]

Names of every InputParam this pipeline expects from execute(**kwargs).

Public alternative to reaching into _inputs from outside the package. Callers that branch on "does this pipeline accept a threshold kwarg" should use this, so the private list shape stays free to evolve.

inputs property

inputs: list[InputParam[Any]]

Every InputParam this pipeline expects, in discovery order.

Public accessor for callers (e.g. the CLI) that need each param's name, type, and default (not just the names from input_names) to build help text or coerce string arguments.

__init__

__init__(
    output: StepResult[Out],
    cache_dir: Path | None = None,
    on_progress: Callable[[Progress], None] | None = None,
    *,
    kind: str | None = None,
    extra_targets: list[StepResult[Any]] | None = None,
    store: CacheStore | None = None,
    metadata: JsonValue = None,
) -> None

Wrap a Deferred graph.

PARAMETER	DESCRIPTION
output	the canonical terminal node of the DAG. execute() without an argument resolves this. TYPE: StepResult[Out]
cache_dir	directory for step results. With neither cache_dir nor store, nothing is cached. TYPE: Path | None DEFAULT: None
store	a CacheStore for step results and BatchStep item caches, used instead of files under cache_dir. TYPE: CacheStore | None DEFAULT: None
on_progress	optional progress callback (passed to ResolveContext). TYPE: Callable[[Progress], None] | None DEFAULT: None
kind	a tag for the pipeline, for example "report/v1". save() writes it to pipeline.json. TYPE: str | None DEFAULT: None
metadata	caller data saved with the pipeline: any JSON value (a string, a number, a list, a dict). save() writes it to pipeline.json and from_directory() reads it back into pipeline.metadata. adag never reads it otherwise. TYPE: JsonValue DEFAULT: None
extra_targets	additional terminal or intermediate nodes whose sub-DAGs should be discovered too. Use this when the pipeline has multiple "ending points" so visualization and resolve() can reach every step in the graph (not just the ones reachable from output). Use Pipeline.from_targets([...]) for a slightly nicer call site when you have no canonical terminal. TYPE: list[StepResult[Any]] | None DEFAULT: None

Source code in src/adag/pipeline.py

def __init__(
    self,
    output: StepResult[Out],
    cache_dir: Path | None = None,
    on_progress: Callable[[Progress], None] | None = None,
    *,
    kind: str | None = None,
    extra_targets: list[StepResult[Any]] | None = None,
    store: CacheStore | None = None,
    metadata: JsonValue = None,
) -> None:
    """Wrap a Deferred graph.

    Args:
        output: the canonical terminal node of the DAG. ``execute()``
            without an argument resolves this.
        cache_dir: directory for step results. With neither ``cache_dir`` nor ``store``,
            nothing is cached.
        store: a ``CacheStore`` for step results and ``BatchStep`` item caches, used
            instead of files under ``cache_dir``.
        on_progress: optional progress callback (passed to ``ResolveContext``).
        kind: a tag for the pipeline, for example ``"report/v1"``. ``save()`` writes it to
            ``pipeline.json``.
        metadata: caller data saved with the pipeline: any JSON value (a string, a number, a
            list, a dict). ``save()`` writes it to ``pipeline.json`` and ``from_directory()``
            reads it back into ``pipeline.metadata``. adag never reads it otherwise.
        extra_targets: additional terminal or intermediate nodes whose
            sub-DAGs should be discovered too. Use this when the
            pipeline has multiple "ending points" so visualization and
            ``resolve()`` can reach every step in the graph (not just
            the ones reachable from ``output``). Use
            ``Pipeline.from_targets([...])`` for a slightly nicer call
            site when you have no canonical terminal.
    """
    self.output = output
    self.cache_dir = cache_dir
    self.store = store
    self.on_progress = on_progress
    self.kind = kind
    if metadata is not None:
        try:
            json.dumps(metadata)
        except (TypeError, ValueError) as exc:
            raise TypeError(f"Pipeline metadata must be a JSON value: {exc}") from exc
    self.metadata: JsonValue = metadata

    self._steps: list[Step[Any]] = []
    self._inputs: list[InputParam[Any]] = []
    self._targets: list[StepResult[Any]] = [output]
    self._discover(output)
    for extra in extra_targets or ():
        if extra is not output and extra not in self._targets:
            self._targets.append(extra)
            self._discover(extra)

from_targets classmethod

from_targets(
    targets: list[StepResult[Any]],
    cache_dir: Path | None = None,
    on_progress: Callable[[Progress], None] | None = None,
    *,
    kind: str | None = None,
    store: CacheStore | None = None,
    metadata: JsonValue = None,
) -> Pipeline[Any]

Build a Pipeline that covers multiple terminal nodes.

Real DAGs often have more than one ending point (e.g. the main result plus a separate audit / dedup artifact). Pass them all here and the Pipeline discovers the union of their sub-graphs. The first target is treated as the canonical output so a zero-arg execute() still has a single thing to return. Downstream callers use resolve(target) to pick any of them.

Source code in src/adag/pipeline.py

@classmethod
def from_targets(
    cls,
    targets: list[StepResult[Any]],
    cache_dir: Path | None = None,
    on_progress: Callable[[Progress], None] | None = None,
    *,
    kind: str | None = None,
    store: CacheStore | None = None,
    metadata: JsonValue = None,
) -> Pipeline[Any]:
    """Build a Pipeline that covers multiple terminal nodes.

    Real DAGs often have more than one ending point (e.g. the main
    result plus a separate audit / dedup artifact). Pass them all
    here and the Pipeline discovers the union of their sub-graphs.
    The first target is treated as the canonical ``output`` so a
    zero-arg ``execute()`` still has a single thing to return.
    Downstream callers use ``resolve(target)`` to pick any of them.
    """
    if not targets:
        raise ValueError("Pipeline.from_targets requires at least one target.")
    head, *rest = targets
    return cls(
        output=head,
        cache_dir=cache_dir,
        on_progress=on_progress,
        kind=kind,
        extra_targets=rest,
        store=store,
        metadata=metadata,
    )

topological_order

topological_order() -> list[Step[Any]]

Steps in a guaranteed topological order (each step after its dependencies).

A named contract (delegates to steps). Consumers that depend on the ordering (deterministic execution, left-to-right layered layout, level grouping) rely on this rather than on the steps discovery detail.

Source code in src/adag/pipeline.py

def topological_order(self) -> list[Step[Any]]:
    """Steps in a guaranteed topological order (each step after its dependencies).

    A named contract (delegates to ``steps``). Consumers that depend on the ordering
    (deterministic execution, left-to-right layered layout, level grouping) rely on this
    rather than on the ``steps`` discovery detail.
    """
    return self.steps

topological_levels

topological_levels() -> list[list[Step[Any]]]

Steps grouped into dependency levels: every step lands one level past its deepest dependency. Steps within a level are mutually independent, so a concurrent executor can run a whole level in parallel; level i+1 starts once level i is resolved.

Pure analysis: no execution change. The foundation for opt-in parallel execution (run each level's steps together) and for correct left-to-right layered layout.

Source code in src/adag/pipeline.py

def topological_levels(self) -> list[list[Step[Any]]]:
    """Steps grouped into dependency levels: every step lands one level past its deepest
    dependency. Steps within a level are mutually independent, so a concurrent executor
    can run a whole level in parallel; level i+1 starts once level i is resolved.

    Pure analysis: no execution change. The foundation for opt-in parallel execution
    (run each level's steps together) and for correct left-to-right layered layout.
    """
    # step -> upstream steps, walked from the targets by StepResult id (no cached map).
    upstreams: defaultdict[Step[Any], set[Step[Any]]] = defaultdict(set)
    seen: set[int] = set()
    stack: list[StepResult[Any]] = list(self._targets)
    while stack:
        node = stack.pop()
        if id(node) in seen:
            continue
        seen.add(id(node))
        for source in node.sources.values():
            if isinstance(source, StepResult):
                upstreams[node.step].add(source.step)
                stack.append(source)

    # self._steps is already topological, so each step's upstreams have a level by now.
    level_of: dict[Step[Any], int] = {}
    levels: list[list[Step[Any]]] = []
    for step in self._steps:
        dependencies = upstreams.get(step, set())
        level = (
            0
            if not dependencies
            else 1 + max(level_of[dependency] for dependency in dependencies)
        )
        level_of[step] = level
        if level == len(levels):
            levels.append([])
        levels[level].append(step)
    return levels

explain_keys

explain_keys(
    params: dict[str, Any] | None = None,
) -> tuple[KeyExplanation, ...]

Explain the cache key of every node, upstream nodes first.

PARAMETER	DESCRIPTION
params	the run's params, as passed to execute. TYPE: dict[str, Any] | None DEFAULT: None

Source code in src/adag/pipeline.py

def explain_keys(self, params: dict[str, Any] | None = None) -> tuple[KeyExplanation, ...]:
    """Explain the cache key of every node, upstream nodes first.

    Args:
        params: the run's params, as passed to ``execute``.
    """
    return tuple(
        node.explain_key(params) for level in self._stepresult_levels() for node in level
    )

validate

validate() -> None

Validate the graph before execute(): every step's required run() input is bound.

Cycles are already rejected at construction (see _discover). This adds the missing-input check: a required parameter of a step's run() that is neither a wired source nor one of the step's dataclass fields would otherwise fail late at execute() as a bare TypeError: run() missing 1 required argument. Opt-in (not run at construction) so a consumer building a graph from dynamic/untrusted input can validate it up front and reject with a 4xx instead of crashing mid-resolution. Raises UnboundInputError naming the step and the unbound params.

Walks the target StepResults directly (visited by id, the same node identity resolve() uses) rather than caching a derived step->sources map.

Source code in src/adag/pipeline.py

def validate(self) -> None:
    """Validate the graph before ``execute()``: every step's required run() input is bound.

    Cycles are already rejected at construction (see ``_discover``). This adds the
    missing-input check: a required parameter of a step's ``run()`` that is neither a
    wired source nor one of the step's dataclass fields would otherwise fail late at
    ``execute()`` as a bare ``TypeError: run() missing 1 required argument``. Opt-in
    (not run at construction) so a consumer building a graph from dynamic/untrusted
    input can validate it up front and reject with a 4xx instead of crashing
    mid-resolution. Raises ``UnboundInputError`` naming the step and the unbound params.

    Walks the target StepResults directly (visited by id, the same node identity
    ``resolve()`` uses) rather than caching a derived step->sources map.
    """
    seen: set[int] = set()
    stack: list[StepResult[Any]] = list(self._targets)
    while stack:
        node = stack.pop()
        if id(node) in seen:
            continue
        seen.add(id(node))
        self._check_inputs_bound(node)
        stack.extend(
            source for source in node.sources.values() if isinstance(source, StepResult)
        )

step_name

step_name(step: Step[Any]) -> str

Directory name for a step: {index}_{name}, falling back to the lowercased class name.

Source code in src/adag/pipeline.py

def step_name(self, step: Step[Any]) -> str:
    """Directory name for a step: {index}_{name}, falling back to the lowercased class name."""
    index = self._steps.index(step)
    width = max(2, math.floor(math.log10(max(len(self._steps), 1))) + 1)
    name = getattr(step, "name", None) or type(step).__name__.lower()
    return f"{index:0{width}d}_{name}"

item_cache

item_cache(namespace: str, schema: type[R]) -> ItemCache[R]

Create an ItemCache in this pipeline's store, or under its cache directory.

Source code in src/adag/pipeline.py

def item_cache(self, namespace: str, schema: type[R]) -> ItemCache[R]:
    """Create an ItemCache in this pipeline's ``store``, or under its cache directory."""
    if self.store is not None:
        return ItemCache(self.cache_dir or Path(), namespace, schema, store=self.store)
    if self.cache_dir is None:
        raise ValueError("Cannot create ItemCache without a cache directory or store.")
    return ItemCache(self.cache_dir, namespace, schema)

execute

execute(
    *,
    observer: PipelineObserver | None = None,
    executor: Executor | None = None,
    dispatcher: BranchDispatcher | None = None,
    parallel: bool = False,
    max_workers: int | None = None,
    **params: Any,
) -> Out

Run the graph and return the result of output.

PARAMETER	DESCRIPTION
observer	receives the run's hooks. Default: LoggingObserver. Pass NullObserver for silence. TYPE: PipelineObserver | None DEFAULT: None
executor	where each step runs. Default: LocalExecutor (this process). TYPE: Executor | None DEFAULT: None
dispatcher	where fan-out branches run. Default: InlineDispatcher (this process). Pass a SubprocessDispatcher or a RemoteDispatcher to run them elsewhere. TYPE: BranchDispatcher | None DEFAULT: None
parallel	run the independent steps of each level on a thread pool. Use it for steps that block on I/O. RunTreeObserver then shows a flat tree. TYPE: bool DEFAULT: False
max_workers	thread pool size for parallel. Default: one thread per step of the level. TYPE: int | None DEFAULT: None
**params	one value per InputParam. TYPE: Any DEFAULT: {}

The five keyword names above are reserved: do not use them as InputParam names.

RAISES	DESCRIPTION
ValueError	a required InputParam has no value.

Source code in src/adag/pipeline.py

def execute(
    self,
    *,
    observer: PipelineObserver | None = None,
    executor: Executor | None = None,
    dispatcher: BranchDispatcher | None = None,
    parallel: bool = False,
    max_workers: int | None = None,
    **params: Any,
) -> Out:
    """Run the graph and return the result of ``output``.

    Args:
        observer: receives the run's hooks. Default: ``LoggingObserver``. Pass ``NullObserver``
            for silence.
        executor: where each step runs. Default: ``LocalExecutor`` (this process).
        dispatcher: where fan-out branches run. Default: ``InlineDispatcher`` (this process).
            Pass a ``SubprocessDispatcher`` or a ``RemoteDispatcher`` to run them elsewhere.
        parallel: run the independent steps of each level on a thread pool. Use it for steps
            that block on I/O. ``RunTreeObserver`` then shows a flat tree.
        max_workers: thread pool size for ``parallel``. Default: one thread per step of the level.
        **params: one value per ``InputParam``.

    The five keyword names above are reserved: do not use them as ``InputParam`` names.

    Raises:
        ValueError: a required ``InputParam`` has no value.
    """
    missing = [p.name for p in self._inputs if p.name not in params and not p.has_default]
    if missing:
        raise ValueError(f"Missing parameters: {missing}")

    context = ResolveContext(
        cache_dir=self.cache_dir,
        store=self.store,
        params=params,
        on_progress=self.on_progress,
        observer=observer or LoggingObserver(),
        executor=executor or LocalExecutor(),
        dispatcher=dispatcher or InlineDispatcher(),
    )
    with observing(context.observer):
        if parallel:
            self._resolve_levels_concurrently(context, max_workers)
        return self.output.resolve(context)

aexecute async

aexecute(
    *,
    observer: PipelineObserver | None = None,
    async_executor: AsyncExecutor | None = None,
    **params: Any,
) -> Out

Run the graph on the running event loop and return the result of output.

AsyncStep bodies are awaited on the loop. Sync step bodies also run on the loop and block it while they run. One step body runs at a time (aresolve_all with max_concurrency=1), so a shared resource such as one database session is never used by two steps at once. Cancelling the task starts no further step.

observer and async_executor are reserved: do not use them as InputParam names. There is no dispatcher: fan-out branches run in this process.

Source code in src/adag/pipeline.py

async def aexecute(
    self,
    *,
    observer: PipelineObserver | None = None,
    async_executor: AsyncExecutor | None = None,
    **params: Any,
) -> Out:
    """Run the graph on the running event loop and return the result of ``output``.

    ``AsyncStep`` bodies are awaited on the loop. Sync step bodies also run on the loop and block
    it while they run. One step body runs at a time (``aresolve_all`` with ``max_concurrency=1``),
    so a shared resource such as one database session is never used by two steps at once.
    Cancelling the task starts no further step.

    ``observer`` and ``async_executor`` are reserved: do not use them as ``InputParam`` names.
    There is no ``dispatcher``: fan-out branches run in this process.
    """
    missing = [p.name for p in self._inputs if p.name not in params and not p.has_default]
    if missing:
        raise ValueError(f"Missing parameters: {missing}")

    context = ResolveContext(
        cache_dir=self.cache_dir,
        store=self.store,
        params=params,
        on_progress=self.on_progress,
        observer=observer or LoggingObserver(),
        async_executor=async_executor or LocalAsyncExecutor(),
    )
    return (await aresolve_all([self.output], context, max_concurrency=1))[0]

resolve

resolve(
    target: StepResult[R],
    *,
    observer: PipelineObserver | None = None,
    executor: Executor | None = None,
    dispatcher: BranchDispatcher | None = None,
    **params: Any,
) -> R

Resolve one node of this pipeline and return its result, from the cache when it is there.

Use it to read an intermediate result after execute(). Do not call step.run() for that: it skips the cache and can return a value that differs from the one downstream steps used.

PARAMETER	DESCRIPTION
target	a node of this pipeline (the object you wired). TYPE: StepResult[R]
**params	the InputParam values, as for execute. TYPE: Any DEFAULT: {}

RAISES	DESCRIPTION
ValueError	target is not part of this pipeline, or a required param is missing.

Source code in src/adag/pipeline.py

def resolve(
    self,
    target: StepResult[R],
    *,
    observer: PipelineObserver | None = None,
    executor: Executor | None = None,
    dispatcher: BranchDispatcher | None = None,
    **params: Any,
) -> R:
    """Resolve one node of this pipeline and return its result, from the cache when it is there.

    Use it to read an intermediate result after ``execute()``. Do not call ``step.run()`` for
    that: it skips the cache and can return a value that differs from the one downstream steps
    used.

    Args:
        target: a node of this pipeline (the object you wired).
        **params: the ``InputParam`` values, as for ``execute``.

    Raises:
        ValueError: ``target`` is not part of this pipeline, or a required param is missing.
    """
    # `_steps` already contains every step reachable from any registered
    # target after _discover ran, so this check covers multi-target
    # pipelines built via from_targets() too.
    if target.step not in self._steps:
        raise ValueError(
            f"target step {type(target.step).__name__} is not part of this pipeline's DAG"
        )
    # Missing-param check is the same as execute() but scoped to the
    # InputParams reachable from `target`. We don't bother pruning:
    # an extra missing-param check is cheap and avoids surprises.
    missing = [p.name for p in self._inputs if p.name not in params and not p.has_default]
    if missing:
        raise ValueError(f"Missing parameters: {missing}")
    context = ResolveContext(
        cache_dir=self.cache_dir,
        store=self.store,
        params=params,
        on_progress=self.on_progress,
        observer=observer or LoggingObserver(),
        executor=executor or LocalExecutor(),
        dispatcher=dispatcher or InlineDispatcher(),
    )
    with observing(context.observer):
        return target.resolve(context)

save

save(path: str | Path) -> None

Save all steps and the pipeline graph to a directory.

Source code in src/adag/pipeline.py

def save(self, path: str | Path) -> None:
    """Save all steps and the pipeline graph to a directory."""
    path = str(path)
    os.makedirs(path, exist_ok=True)

    instance_to_name: dict[int, str] = {}
    step_entries: list[str] = []

    for step in self._steps:
        name = self.step_name(step)
        instance_to_name[id(step)] = name
        step_entries.append(name)

        step_dir = os.path.join(path, name)
        # step.json is owned by Step.save_artifacts (the base writes it via
        # _serialize_step_field; SerializableStep overrides to keep the
        # legacy model_dump format that load_spec() reads). One write per
        # step, no double-write, no Pipeline.save guessing what the spec
        # should look like.
        step.save_artifacts(step_dir)

    edges: list[dict[str, str]] = []
    self._collect_edges(self.output, instance_to_name, edges)

    parameters: dict[str, dict[str, Any]] = {}
    for inp in self._inputs:
        if inp.value_type is not None:
            type_name = inp.value_type.__name__
        else:
            type_name = type(inp.default).__name__ if inp.has_default else "str"
        entry: dict[str, Any] = {"type": type_name}
        if inp.has_default:
            # Reuse the same typed-value coercion as step fields so a default
            # that's a Path / UUID / Enum / BaseModel round-trips cleanly.
            _, default_json = _serialize_step_field(inp.default)
            entry["default"] = default_json if default_json is not None else inp.default
        parameters[inp.name] = entry

    manifest: dict[str, Any] = {"steps": step_entries, "edges": edges, "parameters": parameters}
    if self.kind is not None:
        manifest["kind"] = self.kind
    if self.metadata is not None:
        manifest["metadata"] = self.metadata
    manifest["output"] = instance_to_name[id(self.output.step)]
    with atomic_writer(Path(path).joinpath("pipeline.json")) as f:
        json.dump(manifest, f, indent=2)
        f.write("\n")

    logger.info("Pipeline saved", path=path, steps=len(step_entries))

from_directory classmethod

from_directory(
    path: str | Path,
    step_registry: dict[str, type[Step[Any]]],
    *,
    cache_dir: Path | None = None,
) -> Pipeline

Reconstruct a saved Pipeline from disk.

Reads pipeline.json, loads each step via its class in step_registry (keyed by kind), replays the edge list to rebuild the DAG, and wraps the terminal StepResult.

cache_dir defaults to None: loaded pipelines do not write step results back to disk. Pass cache_dir=Path(path) (or another writable location) when you want the cross-run cache for training or notebook iteration.

Source code in src/adag/pipeline.py

@classmethod
def from_directory(
    cls,
    path: str | Path,
    step_registry: dict[str, type[Step[Any]]],
    *,
    cache_dir: Path | None = None,
) -> Pipeline:
    """Reconstruct a saved Pipeline from disk.

    Reads pipeline.json, loads each step via its class in `step_registry` (keyed by kind),
    replays the edge list to rebuild the DAG, and wraps the terminal StepResult.

    cache_dir defaults to None: loaded pipelines do **not** write step results
    back to disk. Pass cache_dir=Path(path) (or another writable location) when
    you want the cross-run cache for training or notebook iteration.
    """
    base = Path(path)
    with open(base.joinpath("pipeline.json")) as f:
        manifest: dict[str, Any] = json.load(f)

    kind: str | None = manifest.get("kind")
    step_entries: list[str] = list(manifest.get("steps", []))
    edges: list[dict[str, str]] = list(manifest.get("edges", []))
    parameters: dict[str, dict[str, Any]] = dict(manifest.get("parameters", {}))
    output_name: str | None = manifest.get("output")

    inputs: dict[str, InputParam[Any]] = {}
    for param_name, entry in parameters.items():
        if "default" in entry:
            inputs[param_name] = InputParam.with_default(param_name, entry["default"])
        else:
            inputs[param_name] = InputParam(name=param_name)

    loaded_steps: dict[str, Step[Any]] = {}
    for step_name in step_entries:
        step_dir = base.joinpath(step_name)
        with open(step_dir.joinpath("step.json")) as f:
            step_kind: str = json.load(f)["kind"]
        step_cls = step_registry.get(step_kind)
        if step_cls is None:
            known = ", ".join(sorted(step_registry.keys()))
            raise ValueError(f"Unknown step kind {step_kind!r} at {step_dir}. Known: [{known}]")
        loaded_steps[step_name] = step_cls.from_directory(str(step_dir))  # type: ignore[attr-defined]

    results: dict[str, StepResult[Any]] = {}
    for step_name in step_entries:
        step = loaded_steps[step_name]
        inputs_for_step: dict[str, Any] = {}
        for edge in edges:
            target = edge.get("target", "")
            if not target.startswith(f"{step_name}."):
                continue
            input_name = target.split(".", 1)[1]
            source = edge.get("source", "")
            if source.startswith("parameter:"):
                param_name = source.split(":", 1)[1]
                inputs_for_step[input_name] = inputs[param_name]
            else:
                inputs_for_step[input_name] = results[source]
        results[step_name] = step(**inputs_for_step)

    terminal_name = output_name or step_entries[-1]
    output = results[terminal_name]
    return cls(output, cache_dir=cache_dir, kind=kind, metadata=manifest.get("metadata"))

status

status() -> None

Log status of all steps.

Source code in src/adag/pipeline.py

def status(self) -> None:
    """Log status of all steps."""
    for step in self._steps:
        name = self.step_name(step)
        where = self.store if self.store is not None else self.cache_dir
        if step.result_model() is not None and where is not None:
            step_result = self._find_result(step)
            if step_result and step_result.is_cached(where):
                logger.info("step.cached", step=name)
            else:
                logger.info("step.pending", step=name)
        else:
            logger.info("step.ready", step=name)

UnboundInputError

Bases: ValueError

Raised by Pipeline.validate() when a step has a required run() input that is neither wired nor a dataclass field.

Source code in src/adag/pipeline.py

class UnboundInputError(ValueError):
    """Raised by Pipeline.validate() when a step has a required run() input that is neither wired nor a dataclass field."""

SerializableStep dataclass

Bases: Step[Out]

A Step that Pipeline.save writes to disk and Pipeline.from_directory rebuilds.

The contract a subclass implements:

• kind: a non-empty class attribute, for example "my-tokenizer/v1". The registry passed to Pipeline.from_directory maps it back to the class. Raised at class definition if missing.
• spec: an optional dataclass field holding a BaseModel. save_artifacts writes it to step.json.
• from_directory(cls, path): a classmethod that returns the step saved at path. The default rebuilds a step whose fields are spec plus fields with defaults. A step with any other required field must override it, usually with cls(spec=cls.load_spec(path, Spec)). Saving a step that cannot be rebuilt raises TypeError at save time, not at load time.
• save_artifacts(self, path): override only to write extra state (weights, tokenizer files) next to step.json. Call super().save_artifacts(path) first.

Source code in src/adag/serializable.py

class SerializableStep(Step[Out]):
    """A Step that ``Pipeline.save`` writes to disk and ``Pipeline.from_directory`` rebuilds.

    The contract a subclass implements:

    - ``kind``: a non-empty class attribute, for example ``"my-tokenizer/v1"``. The registry passed
      to ``Pipeline.from_directory`` maps it back to the class. Raised at class definition if
      missing.
    - ``spec``: an optional dataclass field holding a ``BaseModel``. ``save_artifacts`` writes it
      to ``step.json``.
    - ``from_directory(cls, path)``: a classmethod that returns the step saved at ``path``. The
      default rebuilds a step whose fields are ``spec`` plus fields with defaults. A step with any
      other required field must override it, usually with ``cls(spec=cls.load_spec(path, Spec))``.
      Saving a step that cannot be rebuilt raises ``TypeError`` at save time, not at load time.
    - ``save_artifacts(self, path)``: override only to write extra state (weights, tokenizer
      files) next to ``step.json``. Call ``super().save_artifacts(path)`` first.
    """

    kind: ClassVar[str]

    def __init_subclass__(cls, **kwargs: Any) -> None:
        """Enforce at class definition time that kind is set and spec is a BaseModel."""
        super().__init_subclass__(**kwargs)
        if "kind" not in cls.__dict__:
            raise TypeError(f"{cls.__name__} must define 'kind'")
        if not cls.kind:
            raise TypeError(f"{cls.__name__}.kind must be a non-empty string")
        if "spec" in inspect.get_annotations(cls):
            try:
                hints = get_type_hints(cls)
                spec_type = hints.get("spec")
                if isinstance(spec_type, type) and not issubclass(spec_type, BaseModel):
                    raise TypeError(
                        f"{cls.__name__}.spec must be annotated with a BaseModel subclass, got {spec_type.__name__}"
                    )
            except NameError:
                pass

    def save_artifacts(self, path: str) -> None:
        """Write step.json with kind and spec (if a spec field exists). Override to add state.

        Atomic via ``atomic_write_text`` so a crash during ``Pipeline.save``
        leaves either the previous step.json intact or the complete new
        one: readers (`from_directory`, `load_spec`) never see a partial
        JSON.
        """
        type(self)._check_loadable()
        os.makedirs(path, exist_ok=True)
        spec_data: dict[str, Any] = {}
        spec_field = getattr(self, "spec", None)
        if isinstance(spec_field, BaseModel):
            spec_data = spec_field.model_dump()
        atomic_write_text(
            Path(path).joinpath("step.json"),
            json.dumps({"kind": self.kind, "spec": spec_data}, indent=2) + "\n",
        )

    @classmethod
    def from_directory(cls, path: str) -> Self:
        """Return the step saved at ``path``. Override when the default cannot rebuild it.

        The default reads ``step.json``, checks ``kind``, and passes ``spec`` (when the class has
        that field) to the constructor. Every other field keeps its default.
        """
        cls._check_loadable()
        hints = get_type_hints(cls)
        spec_type = hints.get("spec")
        if isinstance(spec_type, type) and issubclass(spec_type, BaseModel):
            spec: Any = cls.load_spec(path, spec_type)
            return cls(spec=spec)  # pyright: ignore[reportCallIssue]
        cls.assert_kind(path)
        return cls()

    @classmethod
    def _check_loadable(cls) -> None:
        """Raise ``TypeError`` when the default ``from_directory`` cannot rebuild this class."""
        if "from_directory" in _own_attrs(cls):
            return
        missing = [
            f.name
            for f in dataclasses.fields(cls)  # pyright: ignore[reportArgumentType]
            if f.name != "spec"
            and f.default is dataclasses.MISSING
            and f.default_factory is dataclasses.MISSING
        ]
        if missing:
            raise TypeError(
                f"{cls.__name__} cannot be loaded back: field(s) {', '.join(missing)} have no "
                "default and are not 'spec'. Override from_directory(cls, path) to rebuild it."
            )

    @classmethod
    def load_spec(cls, path: str, spec_type: type[SpecT]) -> SpecT:
        """Read step.json, assert kind matches, and return the validated spec."""
        data = json.loads(Path(path).joinpath("step.json").read_text())
        if data.get("kind") != cls.kind:
            raise ValueError(f"Expected kind {cls.kind!r}, got {data.get('kind')!r}")
        return spec_type.model_validate(data["spec"])

    @classmethod
    def assert_kind(cls, path: str) -> None:
        """Read step.json and assert the kind matches. For steps without a spec."""
        data = json.loads(Path(path).joinpath("step.json").read_text())
        if data.get("kind") != cls.kind:
            raise ValueError(f"Expected kind {cls.kind!r}, got {data.get('kind')!r}")

__init_subclass__

__init_subclass__(**kwargs: Any) -> None

Enforce at class definition time that kind is set and spec is a BaseModel.

Source code in src/adag/serializable.py

def __init_subclass__(cls, **kwargs: Any) -> None:
    """Enforce at class definition time that kind is set and spec is a BaseModel."""
    super().__init_subclass__(**kwargs)
    if "kind" not in cls.__dict__:
        raise TypeError(f"{cls.__name__} must define 'kind'")
    if not cls.kind:
        raise TypeError(f"{cls.__name__}.kind must be a non-empty string")
    if "spec" in inspect.get_annotations(cls):
        try:
            hints = get_type_hints(cls)
            spec_type = hints.get("spec")
            if isinstance(spec_type, type) and not issubclass(spec_type, BaseModel):
                raise TypeError(
                    f"{cls.__name__}.spec must be annotated with a BaseModel subclass, got {spec_type.__name__}"
                )
        except NameError:
            pass

save_artifacts

save_artifacts(path: str) -> None

Write step.json with kind and spec (if a spec field exists). Override to add state.

Atomic via atomic_write_text so a crash during Pipeline.save leaves either the previous step.json intact or the complete new one: readers (from_directory, load_spec) never see a partial JSON.

Source code in src/adag/serializable.py

def save_artifacts(self, path: str) -> None:
    """Write step.json with kind and spec (if a spec field exists). Override to add state.

    Atomic via ``atomic_write_text`` so a crash during ``Pipeline.save``
    leaves either the previous step.json intact or the complete new
    one: readers (`from_directory`, `load_spec`) never see a partial
    JSON.
    """
    type(self)._check_loadable()
    os.makedirs(path, exist_ok=True)
    spec_data: dict[str, Any] = {}
    spec_field = getattr(self, "spec", None)
    if isinstance(spec_field, BaseModel):
        spec_data = spec_field.model_dump()
    atomic_write_text(
        Path(path).joinpath("step.json"),
        json.dumps({"kind": self.kind, "spec": spec_data}, indent=2) + "\n",
    )

from_directory classmethod

from_directory(path: str) -> Self

Return the step saved at path. Override when the default cannot rebuild it.

The default reads step.json, checks kind, and passes spec (when the class has that field) to the constructor. Every other field keeps its default.

Source code in src/adag/serializable.py

@classmethod
def from_directory(cls, path: str) -> Self:
    """Return the step saved at ``path``. Override when the default cannot rebuild it.

    The default reads ``step.json``, checks ``kind``, and passes ``spec`` (when the class has
    that field) to the constructor. Every other field keeps its default.
    """
    cls._check_loadable()
    hints = get_type_hints(cls)
    spec_type = hints.get("spec")
    if isinstance(spec_type, type) and issubclass(spec_type, BaseModel):
        spec: Any = cls.load_spec(path, spec_type)
        return cls(spec=spec)  # pyright: ignore[reportCallIssue]
    cls.assert_kind(path)
    return cls()

load_spec classmethod

load_spec(path: str, spec_type: type[SpecT]) -> SpecT

Read step.json, assert kind matches, and return the validated spec.

Source code in src/adag/serializable.py

@classmethod
def load_spec(cls, path: str, spec_type: type[SpecT]) -> SpecT:
    """Read step.json, assert kind matches, and return the validated spec."""
    data = json.loads(Path(path).joinpath("step.json").read_text())
    if data.get("kind") != cls.kind:
        raise ValueError(f"Expected kind {cls.kind!r}, got {data.get('kind')!r}")
    return spec_type.model_validate(data["spec"])

assert_kind classmethod

assert_kind(path: str) -> None

Read step.json and assert the kind matches. For steps without a spec.

Source code in src/adag/serializable.py

@classmethod
def assert_kind(cls, path: str) -> None:
    """Read step.json and assert the kind matches. For steps without a spec."""
    data = json.loads(Path(path).joinpath("step.json").read_text())
    if data.get("kind") != cls.kind:
        raise ValueError(f"Expected kind {cls.kind!r}, got {data.get('kind')!r}")

WriteJSONL dataclass

Bases: Step[str]

Write the single list[BaseModel] field of a Pydantic result as JSONL.

Reads the upstream Pydantic model, finds its lone list[BaseModel] field, and writes one record per line to path. Returns path so callers can chain a downstream FileSource if they want.

source is typed BaseModel: StepResult._validate is Liskov-aware (accepts any subclass), so a concrete Pydantic result type wires in directly without dropping to Any at the call site.

Raises TypeError if the upstream result has zero or more than one list[BaseModel] fields.

Source code in src/adag/sinks.py

@dataclass(frozen=True)
class WriteJSONL(Step[str]):
    """Write the single ``list[BaseModel]`` field of a Pydantic result as JSONL.

    Reads the upstream Pydantic model, finds its lone ``list[BaseModel]``
    field, and writes one record per line to ``path``. Returns ``path``
    so callers can chain a downstream ``FileSource`` if they want.

    ``source`` is typed ``BaseModel``: ``StepResult._validate`` is
    Liskov-aware (accepts any subclass), so a concrete Pydantic result
    type wires in directly without dropping to ``Any`` at the call site.

    Raises ``TypeError`` if the upstream result has zero or more than one
    ``list[BaseModel]`` fields.
    """

    kind = "adag/write_jsonl/v1"

    path: str

    def run(self, source: BaseModel, **_: Any) -> str:
        list_fields: list[tuple[str, list[BaseModel]]] = []
        for name in source.__class__.model_fields:
            records = _as_basemodel_list(getattr(source, name))
            if records is not None:
                list_fields.append((name, records))

        if len(list_fields) != 1:
            found = [name for name, _ in list_fields]
            raise TypeError(
                f"WriteJSONL requires the source to have exactly one non-empty "
                f"list[BaseModel] field; {type(source).__name__} has "
                f"{len(list_fields)} (found: {found})."
            )

        _field_name, records = list_fields[0]
        del _field_name
        target = Path(self.path)
        with atomic_writer(target) as f:
            for record in records:
                f.write(record.model_dump_json() + "\n")
        return self.path

AsyncStep dataclass

Bases: Step[Out]

A step whose body awaits I/O (a DB session, an HTTP/LLM call). Implement arun (async) instead of run (sync), and resolve it through Pipeline.aexecute / StepResult.aresolve.

An AsyncStep has no synchronous body: it does not override run, so the synchronous path (execute/resolve) hits Step.run's default, which raises a clear message: a sync resolve cannot await an async body. Sync steps are unaffected: their inherited arun just runs run, so a graph can freely MIX sync and async steps and resolve the whole thing with aexecute (each sync step runs inline, each async step is awaited).

Per-request resources (e.g. a SQLAlchemy AsyncSession) flow in like any other input: wire an InputParam and pass it to aexecute. Nothing special is needed on the context.

Source code in src/adag/step.py

class AsyncStep(Step[Out]):
    """A step whose body awaits I/O (a DB session, an HTTP/LLM call). Implement ``arun`` (async) instead of
    ``run`` (sync), and resolve it through ``Pipeline.aexecute`` / ``StepResult.aresolve``.

    An AsyncStep has no synchronous body: it does not override ``run``, so the synchronous path
    (``execute``/``resolve``) hits ``Step.run``'s default, which raises a clear message: a sync
    resolve cannot await an async body. Sync steps are unaffected: their inherited ``arun`` just
    runs ``run``, so a graph can freely MIX sync and async steps and resolve the whole thing with
    ``aexecute`` (each sync step runs inline, each async step is awaited).

    Per-request resources (e.g. a SQLAlchemy ``AsyncSession``) flow in like any other input: wire an
    ``InputParam`` and pass it to ``aexecute``. Nothing special is needed on the context.
    """

    @abstractmethod
    async def arun(self, **kwargs: Any) -> Out:
        """Execute the async step logic. Receives resolved input values as kwargs."""
        ...

arun abstractmethod async

arun(**kwargs: Any) -> Out

Execute the async step logic. Receives resolved input values as kwargs.

Source code in src/adag/step.py

@abstractmethod
async def arun(self, **kwargs: Any) -> Out:
    """Execute the async step logic. Receives resolved input values as kwargs."""
    ...

Step dataclass

Bases: Generic[Out]

Base class of every step: a frozen dataclass with config fields and a run() method.

Fields are part of the cache key, except Config fields. adag passes every field to run() by name, together with the wired inputs.

Class attributes

kind: a stable name for the step. It replaces the class name in the key and in step.json. label, description: text for the graph drawing. result_type: the BaseModel the result is cached as. Inferred from Step[X]. cache_key: a fixed key that replaces the computed hash. strict_content_hash: set to False to skip fields the key cannot encode. pure: False records that a rerun would not give the same value (a model call, a clock read). adag still caches the step and never reads pure. timeout_s: seconds allowed for the async body under aexecute or aresolve. A longer run raises StepTimeout. execute() ignores it, and it cannot stop a blocking sync body. Use a SubprocessDispatcher when you need a hard limit. produces, accepts: optional port names for graph editors. See accepts_from.

Equality is by identity (eq=False), which FanOutStep relies on. A subclass declared with @dataclass(frozen=True) compares by fields.

Source code in src/adag/step.py

@dataclasses.dataclass(frozen=True, eq=False)
class Step(Generic[Out]):
    """Base class of every step: a frozen dataclass with config fields and a ``run()`` method.

    Fields are part of the cache key, except ``Config`` fields. adag passes every field to
    ``run()`` by name, together with the wired inputs.

    Class attributes:
        kind: a stable name for the step. It replaces the class name in the key and in
            ``step.json``.
        label, description: text for the graph drawing.
        result_type: the ``BaseModel`` the result is cached as. Inferred from ``Step[X]``.
        cache_key: a fixed key that replaces the computed hash.
        strict_content_hash: set to ``False`` to skip fields the key cannot encode.
        pure: ``False`` records that a rerun would not give the same value (a model call, a
            clock read). adag still caches the step and never reads ``pure``.
        timeout_s: seconds allowed for the async body under ``aexecute`` or ``aresolve``. A longer
            run raises ``StepTimeout``. ``execute()`` ignores it, and it cannot stop a blocking
            sync body. Use a ``SubprocessDispatcher`` when you need a hard limit.
        produces, accepts: optional port names for graph editors. See ``accepts_from``.

    Equality is by identity (``eq=False``), which ``FanOutStep`` relies on. A subclass declared
    with ``@dataclass(frozen=True)`` compares by fields.
    """

    kind: ClassVar[str] = ""
    label: ClassVar[str] = ""
    description: ClassVar[str] = ""
    result_type: ClassVar[type[BaseModel] | None] = None
    return_type: ClassVar[type[Any] | None] = None
    cache_key: ClassVar[str | None] = None
    strict_content_hash: ClassVar[bool] = True
    pure: ClassVar[bool] = True
    timeout_s: ClassVar[float | None] = None

    # Optional typed ports, read by graph editors to check a connection before it is made.
    # ``produces`` is this step's output port type; ``accepts`` the port types it can consume.
    # ``None`` means untyped: ``accepts_from`` never rejects it. ``Pipeline`` does not check them.
    produces: ClassVar[str | None] = None
    accepts: ClassVar[tuple[str, ...] | None] = None

    def __init_subclass__(cls, **kwargs: Any) -> None:
        super().__init_subclass__(**kwargs)
        overrides_save = _overrides(cls, "save_result")
        overrides_load = _overrides(cls, "load_result")
        if overrides_save != overrides_load:
            raise TypeError(
                f"{cls.__name__} overrides only one of save_result/load_result; override both, "
                "so a cached result can be read back."
            )
        for parent in getattr(cls, "__orig_bases__", []):
            origin = get_origin(parent)
            # Accept both `Step[X]` and any subclass of Step that's also generic
            # (e.g. `SerializableStep[X]`). Without this, `SerializableStep`
            # subclasses never get their return_type / result_type inferred.
            if origin is not None and isinstance(origin, type) and issubclass(origin, Step):
                args = get_args(parent)
                if args and isinstance(args[0], type):
                    if "return_type" not in cls.__dict__:
                        cls.return_type = args[0]
                    if "result_type" not in cls.__dict__ and issubclass(args[0], BaseModel):
                        cls.result_type = args[0]
                return

    def result_model(self) -> type[BaseModel] | None:
        """The concrete pydantic model the result is cached as (JSON round-trip). Defaults to the class
        ``result_type`` inferred from ``Step[X]``; override when the result type is generic (e.g. ``Dataset[T]``)
        so the parametrized model is known at cache-load time, not the unparametrized base.
        """
        return type(self).result_type

    def output_type(self) -> type[Any] | None:
        """The deferred output type used for wiring / connection checks. Defaults to the class ``return_type``;
        override alongside ``result_model`` when the result type is generic.
        """
        return type(self).return_type

    def save_result(self, result: Out, path: Path) -> None:
        """Write ``result`` to ``path``. Override with ``load_result`` for a non-JSON result.

        ``path`` does not exist yet. Create it as a file or as a directory (a directory holds a
        result plus any sidecar files). adag passes a temporary path and renames it into place
        when this returns, so a crash never leaves a half-written result. The default writes the
        ``BaseModel`` result as JSON.
        """
        if not isinstance(result, BaseModel):
            raise TypeError(
                f"{type(self).__name__}: the default save_result writes a BaseModel; got "
                f"{type(result).__name__}. Override save_result and load_result."
            )
        atomic_write_text(path, result.model_dump_json(indent=2))

    def load_result(self, path: Path) -> Out:
        """Read the result that ``save_result`` wrote at ``path``.

        Raise any exception when the entry is not valid: adag treats it as a cache miss and runs
        the step again. The default reads JSON with ``result_model()``.
        """
        model = self.result_model()
        if model is None:
            raise TypeError(f"{type(self).__name__} has no BaseModel result type to load")
        return model.model_validate_json(path.read_text())  # pyright: ignore[reportReturnType]

    @classmethod
    def caches_to_path(cls) -> bool:
        """Return whether this step overrides ``save_result``/``load_result``.

        Such a step is cached under ``cache_dir`` as a file or directory, never in a ``store``.
        """
        return _overrides(cls, "save_result")

    def accepts_from(self, producer: Step[Any] | str) -> bool:
        """Return whether this step can consume ``producer``'s output port.

        ``producer`` is a step (its ``produces`` is read) or a port name. Returns ``True`` when
        either side declares no port; otherwise ``producer.produces`` must be in ``accepts``.
        """
        produces = producer if isinstance(producer, str) else producer.produces
        if produces is None or self.accepts is None:
            return True
        return produces in self.accepts

    def __call__(self, **kwargs: Any) -> StepResult[Out]:
        """Wire this step into the DAG. Returns a StepResult that chains downstream.

        Warns ``AdagDeprecationWarning`` when an input has the name of a field that holds a ``Step``:
        the field copy is hashed but never run, and the input value is what ``run()`` gets. The
        first release documented this pattern, so it still works. It becomes a ``TypeError`` in
        adag 1.0. Wire the upstream step once, as an input.

        Raises:
            TypeError: a field holds a ``StepResult``. A field is part of the key, not an edge;
                pass the upstream node as a call input instead.
        """
        for f in dataclasses.fields(self):
            if isinstance(getattr(self, f.name), StepResult):
                raise TypeError(
                    f"{type(self).__name__}.{f.name} holds a StepResult. A field is hashed into "
                    f"the cache key, not wired as an edge: pass it as a call input, "
                    f"{type(self).__name__}(...)({f.name}=node)."
                )
            if f.name in kwargs and isinstance(getattr(self, f.name), Step):
                warnings.warn(
                    f"{type(self).__name__}.{f.name} is both a Step field and an input. "
                    f"Remove the field and pass the upstream result as {f.name}=... only. "
                    "This becomes a TypeError in adag 1.0.",
                    AdagDeprecationWarning,
                    stacklevel=2,
                )
        return StepResult(step=self, sources=kwargs)

    def run(self, **kwargs: Any) -> Out:
        """Execute the step's synchronous logic; receives resolved inputs as kwargs.

        Override in a synchronous step. Deliberately NOT ``@abstractmethod``: an
        ``AsyncStep`` (implements ``arun``) and a ``FanOutStep`` (driven via ``execute``)
        have no synchronous ``run``, and marking it abstract would make them
        un-instantiable. The default raise instead guards the real misuse cases, a sync
        step that forgot ``run``, or a sync resolve of an async step.
        """
        del kwargs
        raise NotImplementedError(
            f"{type(self).__name__} has no run(): a synchronous Step must implement run(); "
            "an AsyncStep implements arun() and must be resolved via Pipeline.aexecute()."
        )

    def execute(self, context: ResolveContext, /, **kwargs: Any) -> Out:
        """Framework entry point. Override if the step needs per-call state
        (e.g. BatchStep needs context.cache_dir for its ItemCache). Most
        steps don't: the default binds the run's observer as the ambient
        ``current_observer()`` (so step internals, incl. worker threads and
        asyncio tasks, can emit observer hooks) and delegates to run().
        """
        with observing(context.observer):
            return self.run(**kwargs)

    async def arun(self, **kwargs: Any) -> Out:
        """Async step body. Default delegates to the sync ``run`` so a sync step works unchanged on the
        async path; a step that awaits I/O subclasses ``AsyncStep`` and overrides this.
        """
        return self.run(**kwargs)

    async def aexecute(self, context: ResolveContext, /, **kwargs: Any) -> Out:
        """Async framework entry point: the awaitable mirror of ``execute``. Override for per-call state;
        the default delegates to ``arun``.
        """
        del context
        return await self.arun(**kwargs)

    def content_hash(self) -> str | None:
        """Return the step's own hash (class, code and fields), or ``None`` when a field blocks it.

        Identity is ``kind`` when set, else the class name. The class source is hashed in both cases,
        so a code edit changes the hash. Returns ``cache_key`` unchanged when the class sets it.

        - ``Config`` fields: skipped.
        - ``FileSource`` fields: hashed by file content. A missing file blocks the hash.
        - A field that holds a ``Step``: hashed by that step's hash.
        - Other values: encoded by ``adag.hashing.encode_value``.

        Raises:
            UnhashableValueError: a field has no stable encoding and ``strict_content_hash`` is True.
        """
        pinned = type(self).cache_key
        if pinned is not None:
            return pinned

        h = hashlib.sha256()
        for _, _, _, chunk in self._key_chunks():
            if chunk is None:
                return None
            h.update(chunk)
        return h.hexdigest()[:12]

    def _key_chunks(self) -> Iterator[tuple[ComponentKind, str, Any, bytes | None]]:
        """Yield each component of this step's key: ``(kind, name, value, bytes)``.

        ``content_hash`` feeds the bytes to sha256 in this order; ``explain_key`` reads the same
        tuples. ``None`` bytes block the key. Empty bytes add nothing.
        """
        kind = type(self).kind
        if kind:
            yield ComponentKind.IDENTITY, "", kind, b"KIND:" + encode_value(kind)
        else:
            name = type(self).__name__
            yield ComponentKind.IDENTITY, "", name, b"STEP:" + encode_value(name)
        # Source is hashed for kind-tagged steps too: editing a step's code must
        # invalidate its results whether or not it has a serialization kind.
        code_bytes = class_source_hash(type(self))
        if code_bytes:
            yield ComponentKind.CODE, "", code_bytes.hex()[:12], b"CODE:" + code_bytes

        for field in dataclasses.fields(self):
            value = getattr(self, field.name)
            if field.name == "node_label":
                # Display-only label for the graph renderer, not compute identity.
                yield ComponentKind.SKIPPED, field.name, value, b""
                continue
            chunk = self._hash_field(field.name, value)
            if isinstance(value, Config) or chunk == b"":
                component = ComponentKind.SKIPPED
            elif isinstance(value, FileSource):
                component = ComponentKind.FILE
            elif isinstance(value, Step):
                component = ComponentKind.NESTED_STEP
            else:
                component = ComponentKind.FIELD
            yield component, field.name, value, chunk

    def _hash_field(self, field_name: str, value: Any) -> bytes | None:
        """Hash a single dataclass field for content_hash.

        Returns:
          - `bytes` to mix into the parent hash (empty bytes = skip).
          - `None` to block the parent hash entirely (unsatisfied input).

        Raises:
            UnhashableValueError: the field type has no stable encoding. Wrap it in
                ``Config(...)``, or set ``strict_content_hash = False`` to skip it.
        """
        name = encode_value(field_name)
        if isinstance(value, Config):
            return b""
        if isinstance(value, FileSource):
            file_content_hash = value.content_hash()
            if file_content_hash is None:
                return None
            return b"FILE:" + name + encode_value(file_content_hash)
        if isinstance(value, Step):
            sub = value.content_hash()
            if sub is None:
                return None
            return b"SUB:" + name + encode_value(sub)
        try:
            return (
                b"VALUE:"
                + name
                + encode_value(value, f"field {field_name!r} of {type(self).__name__}")
            )
        except UnhashableValueError:
            if type(self).strict_content_hash:
                raise
            return b""

    def save_artifacts(self, path: str) -> None:
        """Write step.json with the step's kind and serialized config fields.

        Default implementation captures the dataclass fields that ``_serialize_step_field``
        recognises (scalars, Path, UUID, Enum, BaseModel, composites thereof).
        Fields representing runtime state (Step instances, tokenizers, HF clients,
        model handles) are skipped: those subclasses override this method, call
        ``super().save_artifacts(path)``, then write their own state files
        (safetensors, tokenizer files, etc.) next to step.json.

        SerializableStep overrides to keep the legacy ``data["spec"] = model_dump``
        format that ``load_spec()`` expects; the formats stay distinct because
        flattening would break round-trip for every existing saved pipeline.
        """
        os.makedirs(path, exist_ok=True)
        step_config: dict[str, Any] = {}
        for field in dataclasses.fields(self):
            handled, json_value = _serialize_step_field(getattr(self, field.name))
            if handled:
                step_config[field.name] = json_value

        kind = getattr(self, "kind", None) or type(self).__name__.lower()
        step_json: dict[str, Any] = {"kind": kind, "spec": step_config}
        atomic_write_text(Path(path).joinpath("step.json"), json.dumps(step_json, indent=2) + "\n")

    def load_artifacts(self, path: str) -> None:
        """Load step-specific artifacts from a directory.

        Override in steps that carry heavy state (neural network modules, etc.).
        """
        del path

result_model

result_model() -> type[BaseModel] | None

The concrete pydantic model the result is cached as (JSON round-trip). Defaults to the class result_type inferred from Step[X]; override when the result type is generic (e.g. Dataset[T]) so the parametrized model is known at cache-load time, not the unparametrized base.

Source code in src/adag/step.py

def result_model(self) -> type[BaseModel] | None:
    """The concrete pydantic model the result is cached as (JSON round-trip). Defaults to the class
    ``result_type`` inferred from ``Step[X]``; override when the result type is generic (e.g. ``Dataset[T]``)
    so the parametrized model is known at cache-load time, not the unparametrized base.
    """
    return type(self).result_type

output_type

output_type() -> type[Any] | None

The deferred output type used for wiring / connection checks. Defaults to the class return_type; override alongside result_model when the result type is generic.

Source code in src/adag/step.py

def output_type(self) -> type[Any] | None:
    """The deferred output type used for wiring / connection checks. Defaults to the class ``return_type``;
    override alongside ``result_model`` when the result type is generic.
    """
    return type(self).return_type

save_result

save_result(result: Out, path: Path) -> None

Write result to path. Override with load_result for a non-JSON result.

path does not exist yet. Create it as a file or as a directory (a directory holds a result plus any sidecar files). adag passes a temporary path and renames it into place when this returns, so a crash never leaves a half-written result. The default writes the BaseModel result as JSON.

Source code in src/adag/step.py

def save_result(self, result: Out, path: Path) -> None:
    """Write ``result`` to ``path``. Override with ``load_result`` for a non-JSON result.

    ``path`` does not exist yet. Create it as a file or as a directory (a directory holds a
    result plus any sidecar files). adag passes a temporary path and renames it into place
    when this returns, so a crash never leaves a half-written result. The default writes the
    ``BaseModel`` result as JSON.
    """
    if not isinstance(result, BaseModel):
        raise TypeError(
            f"{type(self).__name__}: the default save_result writes a BaseModel; got "
            f"{type(result).__name__}. Override save_result and load_result."
        )
    atomic_write_text(path, result.model_dump_json(indent=2))

load_result

load_result(path: Path) -> Out

Read the result that save_result wrote at path.

Raise any exception when the entry is not valid: adag treats it as a cache miss and runs the step again. The default reads JSON with result_model().

Source code in src/adag/step.py

def load_result(self, path: Path) -> Out:
    """Read the result that ``save_result`` wrote at ``path``.

    Raise any exception when the entry is not valid: adag treats it as a cache miss and runs
    the step again. The default reads JSON with ``result_model()``.
    """
    model = self.result_model()
    if model is None:
        raise TypeError(f"{type(self).__name__} has no BaseModel result type to load")
    return model.model_validate_json(path.read_text())  # pyright: ignore[reportReturnType]

caches_to_path classmethod

caches_to_path() -> bool

Return whether this step overrides save_result/load_result.

Such a step is cached under cache_dir as a file or directory, never in a store.

Source code in src/adag/step.py

@classmethod
def caches_to_path(cls) -> bool:
    """Return whether this step overrides ``save_result``/``load_result``.

    Such a step is cached under ``cache_dir`` as a file or directory, never in a ``store``.
    """
    return _overrides(cls, "save_result")

accepts_from

accepts_from(producer: Step[Any] | str) -> bool

Return whether this step can consume producer's output port.

producer is a step (its produces is read) or a port name. Returns True when either side declares no port; otherwise producer.produces must be in accepts.

Source code in src/adag/step.py

def accepts_from(self, producer: Step[Any] | str) -> bool:
    """Return whether this step can consume ``producer``'s output port.

    ``producer`` is a step (its ``produces`` is read) or a port name. Returns ``True`` when
    either side declares no port; otherwise ``producer.produces`` must be in ``accepts``.
    """
    produces = producer if isinstance(producer, str) else producer.produces
    if produces is None or self.accepts is None:
        return True
    return produces in self.accepts

__call__

__call__(**kwargs: Any) -> StepResult[Out]

Wire this step into the DAG. Returns a StepResult that chains downstream.

Warns AdagDeprecationWarning when an input has the name of a field that holds a Step: the field copy is hashed but never run, and the input value is what run() gets. The first release documented this pattern, so it still works. It becomes a TypeError in adag 1.0. Wire the upstream step once, as an input.

RAISES	DESCRIPTION
TypeError	a field holds a StepResult. A field is part of the key, not an edge; pass the upstream node as a call input instead.

Source code in src/adag/step.py

def __call__(self, **kwargs: Any) -> StepResult[Out]:
    """Wire this step into the DAG. Returns a StepResult that chains downstream.

    Warns ``AdagDeprecationWarning`` when an input has the name of a field that holds a ``Step``:
    the field copy is hashed but never run, and the input value is what ``run()`` gets. The
    first release documented this pattern, so it still works. It becomes a ``TypeError`` in
    adag 1.0. Wire the upstream step once, as an input.

    Raises:
        TypeError: a field holds a ``StepResult``. A field is part of the key, not an edge;
            pass the upstream node as a call input instead.
    """
    for f in dataclasses.fields(self):
        if isinstance(getattr(self, f.name), StepResult):
            raise TypeError(
                f"{type(self).__name__}.{f.name} holds a StepResult. A field is hashed into "
                f"the cache key, not wired as an edge: pass it as a call input, "
                f"{type(self).__name__}(...)({f.name}=node)."
            )
        if f.name in kwargs and isinstance(getattr(self, f.name), Step):
            warnings.warn(
                f"{type(self).__name__}.{f.name} is both a Step field and an input. "
                f"Remove the field and pass the upstream result as {f.name}=... only. "
                "This becomes a TypeError in adag 1.0.",
                AdagDeprecationWarning,
                stacklevel=2,
            )
    return StepResult(step=self, sources=kwargs)

run

run(**kwargs: Any) -> Out

Execute the step's synchronous logic; receives resolved inputs as kwargs.

Override in a synchronous step. Deliberately NOT @abstractmethod: an AsyncStep (implements arun) and a FanOutStep (driven via execute) have no synchronous run, and marking it abstract would make them un-instantiable. The default raise instead guards the real misuse cases, a sync step that forgot run, or a sync resolve of an async step.

Source code in src/adag/step.py

def run(self, **kwargs: Any) -> Out:
    """Execute the step's synchronous logic; receives resolved inputs as kwargs.

    Override in a synchronous step. Deliberately NOT ``@abstractmethod``: an
    ``AsyncStep`` (implements ``arun``) and a ``FanOutStep`` (driven via ``execute``)
    have no synchronous ``run``, and marking it abstract would make them
    un-instantiable. The default raise instead guards the real misuse cases, a sync
    step that forgot ``run``, or a sync resolve of an async step.
    """
    del kwargs
    raise NotImplementedError(
        f"{type(self).__name__} has no run(): a synchronous Step must implement run(); "
        "an AsyncStep implements arun() and must be resolved via Pipeline.aexecute()."
    )

execute

execute(context: ResolveContext, /, **kwargs: Any) -> Out

Framework entry point. Override if the step needs per-call state (e.g. BatchStep needs context.cache_dir for its ItemCache). Most steps don't: the default binds the run's observer as the ambient current_observer() (so step internals, incl. worker threads and asyncio tasks, can emit observer hooks) and delegates to run().

Source code in src/adag/step.py

def execute(self, context: ResolveContext, /, **kwargs: Any) -> Out:
    """Framework entry point. Override if the step needs per-call state
    (e.g. BatchStep needs context.cache_dir for its ItemCache). Most
    steps don't: the default binds the run's observer as the ambient
    ``current_observer()`` (so step internals, incl. worker threads and
    asyncio tasks, can emit observer hooks) and delegates to run().
    """
    with observing(context.observer):
        return self.run(**kwargs)

arun async

arun(**kwargs: Any) -> Out

Async step body. Default delegates to the sync run so a sync step works unchanged on the async path; a step that awaits I/O subclasses AsyncStep and overrides this.

Source code in src/adag/step.py

async def arun(self, **kwargs: Any) -> Out:
    """Async step body. Default delegates to the sync ``run`` so a sync step works unchanged on the
    async path; a step that awaits I/O subclasses ``AsyncStep`` and overrides this.
    """
    return self.run(**kwargs)

aexecute async

aexecute(context: ResolveContext, /, **kwargs: Any) -> Out

Async framework entry point: the awaitable mirror of execute. Override for per-call state; the default delegates to arun.

Source code in src/adag/step.py

async def aexecute(self, context: ResolveContext, /, **kwargs: Any) -> Out:
    """Async framework entry point: the awaitable mirror of ``execute``. Override for per-call state;
    the default delegates to ``arun``.
    """
    del context
    return await self.arun(**kwargs)

content_hash

content_hash() -> str | None

Return the step's own hash (class, code and fields), or None when a field blocks it.

Identity is kind when set, else the class name. The class source is hashed in both cases, so a code edit changes the hash. Returns cache_key unchanged when the class sets it.

• Config fields: skipped.
• FileSource fields: hashed by file content. A missing file blocks the hash.
• A field that holds a Step: hashed by that step's hash.
• Other values: encoded by adag.hashing.encode_value.

RAISES	DESCRIPTION
UnhashableValueError	a field has no stable encoding and strict_content_hash is True.

Source code in src/adag/step.py

def content_hash(self) -> str | None:
    """Return the step's own hash (class, code and fields), or ``None`` when a field blocks it.

    Identity is ``kind`` when set, else the class name. The class source is hashed in both cases,
    so a code edit changes the hash. Returns ``cache_key`` unchanged when the class sets it.

    - ``Config`` fields: skipped.
    - ``FileSource`` fields: hashed by file content. A missing file blocks the hash.
    - A field that holds a ``Step``: hashed by that step's hash.
    - Other values: encoded by ``adag.hashing.encode_value``.

    Raises:
        UnhashableValueError: a field has no stable encoding and ``strict_content_hash`` is True.
    """
    pinned = type(self).cache_key
    if pinned is not None:
        return pinned

    h = hashlib.sha256()
    for _, _, _, chunk in self._key_chunks():
        if chunk is None:
            return None
        h.update(chunk)
    return h.hexdigest()[:12]

save_artifacts

save_artifacts(path: str) -> None

Write step.json with the step's kind and serialized config fields.

Default implementation captures the dataclass fields that _serialize_step_field recognises (scalars, Path, UUID, Enum, BaseModel, composites thereof). Fields representing runtime state (Step instances, tokenizers, HF clients, model handles) are skipped: those subclasses override this method, call super().save_artifacts(path), then write their own state files (safetensors, tokenizer files, etc.) next to step.json.

SerializableStep overrides to keep the legacy data["spec"] = model_dump format that load_spec() expects; the formats stay distinct because flattening would break round-trip for every existing saved pipeline.

Source code in src/adag/step.py

def save_artifacts(self, path: str) -> None:
    """Write step.json with the step's kind and serialized config fields.

    Default implementation captures the dataclass fields that ``_serialize_step_field``
    recognises (scalars, Path, UUID, Enum, BaseModel, composites thereof).
    Fields representing runtime state (Step instances, tokenizers, HF clients,
    model handles) are skipped: those subclasses override this method, call
    ``super().save_artifacts(path)``, then write their own state files
    (safetensors, tokenizer files, etc.) next to step.json.

    SerializableStep overrides to keep the legacy ``data["spec"] = model_dump``
    format that ``load_spec()`` expects; the formats stay distinct because
    flattening would break round-trip for every existing saved pipeline.
    """
    os.makedirs(path, exist_ok=True)
    step_config: dict[str, Any] = {}
    for field in dataclasses.fields(self):
        handled, json_value = _serialize_step_field(getattr(self, field.name))
        if handled:
            step_config[field.name] = json_value

    kind = getattr(self, "kind", None) or type(self).__name__.lower()
    step_json: dict[str, Any] = {"kind": kind, "spec": step_config}
    atomic_write_text(Path(path).joinpath("step.json"), json.dumps(step_json, indent=2) + "\n")

load_artifacts

load_artifacts(path: str) -> None

Load step-specific artifacts from a directory.

Override in steps that carry heavy state (neural network modules, etc.).

Source code in src/adag/step.py

def load_artifacts(self, path: str) -> None:
    """Load step-specific artifacts from a directory.

    Override in steps that carry heavy state (neural network modules, etc.).
    """
    del path

StepResult

Bases: Deferred[Out]

Deferred result of a Step. Handles caching, upstream resolution, and type validation.

Source code in src/adag/step.py

class StepResult(Deferred[Out]):
    """Deferred result of a Step. Handles caching, upstream resolution, and type validation."""

    def __init__(self, step: Step[Out], sources: dict[str, _SourceValue]) -> None:
        self.step = step
        self.sources: dict[str, _SourceValue] = sources
        self.deferred_type = step.output_type()
        # Memoize content_hash(): it recurses into every upstream and is called many times
        # per node (resolve, is_cached, load), so a deep chain would recompute hashes
        # ~quadratically without this. Single slot, keyed on the params object's IDENTITY
        # (graph + params are immutable within one execute(); a new params object recomputes).
        # _hash_computed separates "not computed yet" from "computed as None" (a blocked
        # step hashes to None), so params=None still computes exactly once.
        self._hash_computed: bool = False
        self._hash_params: dict[str, Any] | None = None
        self._hash_value: str | None = None
        self._validate()

    def _validate(self) -> None:
        """Runtime type check at wiring time. Best-effort: skips if hints fail. An ``AsyncStep``'s input
        types live on ``arun`` (it has no synchronous ``run``), so validate against the step's real body.
        """
        target = type(self.step).arun if isinstance(self.step, AsyncStep) else type(self.step).run
        try:
            run_hints = get_type_hints(target)
        except Exception as exc:
            # Best-effort: a forward ref that can't resolve (missing import, stringized
            # annotation) means we can't type-check the wiring. Log it rather than fail
            # silently, then skip (wiring still works; this is a guard, not a gate).
            logger.debug(
                "step.validate.hints_unavailable", step=type(self.step).__name__, error=str(exc)
            )
            return
        for name, source in self.sources.items():
            if isinstance(source, Deferred):
                expected: Any = run_hints.get(name)
                actual: Any = source.deferred_type
                if expected and actual and not _types_compatible(expected, actual):
                    expected_name = getattr(expected, "__name__", repr(expected))
                    actual_name = getattr(actual, "__name__", repr(actual))
                    raise TypeError(
                        f"Type mismatch: {type(source).__name__} produces {actual_name}, "
                        f"but {type(self.step).__name__} expects {expected_name} for '{name}'"
                    )

    def _memo_get(self, context: ResolveContext) -> Out | _Miss:
        """This node's memoized result in ``context``, or ``_CACHE_MISS``. Hold ``_lock``.

        Entries are ``(node, value)`` keyed by ``id(node)``. Holding the node keeps its id from
        being reused while the entry exists, and the identity check rejects a stale entry.
        """
        entry = context._resolved.get(id(self))
        if entry is not None and entry[0] is self:
            return entry[1]
        return _CACHE_MISS

    def _check_cache(self, context: ResolveContext) -> Out | _Miss:
        """The shared, synchronous *front* of ``resolve``/``aresolve``: per-execute memo
        lookup, ``on_step_start``, and the disk-cache hit path. Returns the resolved value
        on a memo/cache hit, or ``_CACHE_MISS`` if the step still has to run.

        Disk IO is synchronous on both paths. It is not the bottleneck and keeps the cache
        format byte-identical. ``context._lock`` guards the memo dict so the parallel
        executor's concurrent nodes are safe regardless of the GIL.
        """
        # Per-execute memoization, by id(). The memo holds the node too (see _memo_get), so a
        # context that outlives its graph cannot hand one node's result to another.
        with context._lock:
            memo = self._memo_get(context)
        if not isinstance(memo, _Miss):
            return memo

        step_name = type(self.step).__name__
        # Announce entry BEFORE the recursive hash/cache check, so progress shows the instant
        # a step is reached. Firing on_step_start before source resolution also brackets every
        # child's start/finish inside this step's. That nesting is what the RunTreeObserver
        # turns back into the execution tree.
        context.observer.on_step_start(step_name)

        if self.step.caches_to_path():
            loaded = self._try_load_path(self._result_dir(context), context.params)
        else:
            store = context.result_store()
            loaded = self._try_load(store, params=context.params) if store is not None else None
        if loaded is not None:
            if context.on_progress:
                context.on_progress(Progress(step=step_name, message="cache hit"))
            with context._lock:
                context._resolved[id(self)] = (self, loaded)
            context.observer.on_step_cached(step_name)
            return loaded

        if context.on_progress:
            context.on_progress(Progress(step=step_name, message="resolving"))
        return _CACHE_MISS

    def _finalize(self, result: Out, context: ResolveContext, started_at: float) -> Out:
        """The shared, synchronous *back* of ``resolve``/``aresolve``: observer-finish,
        disk-cache write, and the memo write. Same on both paths.
        """
        step_name = type(self.step).__name__
        context.observer.on_step_finished(step_name, time.perf_counter() - started_at)
        if self.step.caches_to_path():
            cache_dir = self._result_dir(context)
            if cache_dir is not None:
                result = self._save_path(result, cache_dir, context.params)
                emit("step.saved", step=step_name, key=self.content_hash(params=context.params))
                self._write_key_record(context)
        else:
            store = context.result_store()
            if store is not None and self.step.result_model() is not None:
                self._save(result, store, params=context.params)
                emit("step.saved", step=step_name, key=self.content_hash(params=context.params))
                self._write_key_record(context)
        with context._lock:
            context._resolved[id(self)] = (self, result)
        return result

    def _write_key_record(self, context: ResolveContext) -> None:
        """Write this node's key explanation to ``.keys/<key>.json`` in the result store.

        The record lets a later release compare a new key with the one a result was saved
        under. A failed write never fails the run: it emits ``step.key_record.failed``.
        """
        store = context.result_store()
        key = self.content_hash(params=context.params)
        if store is None or key is None:
            return
        try:
            from adag import __version__  # late: adag/__init__ imports this module

            record = {
                "record_version": 1,
                "adag_version": __version__,
                "saved_at": datetime.now(UTC).isoformat(),
                **self.explain_key(context.params).redacted().to_dict(),
            }
            store.set(f"{KEY_RECORD_PREFIX}{key}.json", json.dumps(record, indent=1))
        except Exception as exc:
            emit("step.key_record.failed", step=type(self.step).__name__, key=key, error=str(exc))

    def resolve(self, context: ResolveContext) -> Out:
        """Resolve upstream sources, run the step, cache if applicable.

        Memoizes by ``id(self)`` so multiple downstream consumers of the SAME ``StepResult``
        share one computed result within a ``Pipeline.execute``. This is critical for non-cached
        steps returning non-serializable objects (e.g. model handles), where consumers must
        see the same instance, not two re-executions. The cache/memo/observer skeleton lives
        in ``_check_cache``/``_finalize`` (shared with ``aresolve``); only the two lines that
        genuinely differ (upstream resolution and the step body) stay here, run inline.

        In-flight dedup makes that guarantee hold under CONCURRENCY too: the memo only
        collapses resolutions that do not overlap in time, so N branch workers reaching one
        shared upstream at once would each miss the memo and run the body N times. Here the
        first thread to miss the memo claims ownership (an ``Event`` in ``context._inflight``);
        concurrent threads find that Event, wait on it, then loop: a success left the value in
        the memo (returned), a failure removed the marker so they re-claim and resolve it
        themselves. Uncontended (single-threaded) there is never a waiter, so behaviour and
        cost are unchanged bar creating one Event that is set immediately.
        """
        while True:
            with context._lock:
                memo = self._memo_get(context)
                if not isinstance(memo, _Miss):
                    return memo
                in_flight = context._inflight.get(id(self))
                if in_flight is None:
                    own_event = Event()
                    context._inflight[id(self)] = own_event
                    break
            # Another thread owns this exact node's resolution. Wait for it, then loop.
            in_flight.wait()
        try:
            cached = self._check_cache(context)
            if not isinstance(cached, _Miss):
                return cached
            resolved = self._merge_step_defaults(self._resolve_sources(context))
            started_at = time.perf_counter()
            result = context.executor.run(self.step, context, resolved)
            return self._finalize(result, context, started_at)
        finally:
            with context._lock:
                context._inflight.pop(id(self), None)
            own_event.set()

    def _merge_step_defaults(self, resolved: dict[str, Any]) -> dict[str, Any]:
        """Merge the step's own (non-source) dataclass field values into the
        resolved sources, unwrapping Config fields so run() receives the
        underlying value. Resolved sources win over default field values.
        """
        fields: dict[str, Any] = {}
        for f in dataclasses.fields(self.step):
            if f.name in resolved:
                continue
            value: Any = getattr(self.step, f.name)
            if isinstance(value, Config):
                value = value.value
            fields[f.name] = value
        return {**fields, **resolved}

    def _resolve_sources(self, context: ResolveContext) -> dict[str, Any]:
        """Resolve all upstream sources."""
        resolved: dict[str, Any] = {}
        for name, source in self.sources.items():
            if isinstance(source, Config):
                resolved[name] = source.value
            elif isinstance(source, Deferred):
                resolved[name] = source.resolve(context)
            else:
                resolved[name] = source
        return resolved

    async def aresolve(self, context: ResolveContext) -> Out:
        """Resolve this node on the event loop and return its result. Public API.

        Use it to schedule async steps yourself, for example one ``asyncio.gather`` per
        topological level inside an async request handler. Pass the same ``ResolveContext`` to
        every call: it memoizes each node, so a shared upstream runs once and a later call reuses
        earlier results. Upstream nodes are awaited, async step bodies run on the loop, and sync
        bodies run unchanged. A cancelled task starts no further step.

        Gathered calls that reach the same node at the same time run its body once: the first
        task claims the node and the others wait for it. If that body raises, a waiting task
        resolves the node itself.
        """
        while True:
            with context._lock:
                memo = self._memo_get(context)
                if not isinstance(memo, _Miss):
                    return memo
                in_flight = context._ainflight.get(id(self))
                if in_flight is None:
                    own = asyncio.get_running_loop().create_future()
                    context._ainflight[id(self)] = own
                    break
            # Another task owns this node. Wait without taking its exception, then loop: a
            # success left the value in the memo, a failure removed the marker.
            await asyncio.wait([in_flight])
        try:
            return await self._aresolve_owned(context)
        finally:
            with context._lock:
                context._ainflight.pop(id(self), None)
            if not own.done():
                own.set_result(None)

    async def _aresolve_owned(self, context: ResolveContext) -> Out:
        """Resolve this node on the loop, once this task owns it (see ``aresolve``)."""
        cached = self._check_cache(context)
        if not isinstance(cached, _Miss):
            return cached
        resolved = self._merge_step_defaults(await self._aresolve_sources(context))
        started_at = time.perf_counter()
        timeout = self.step.timeout_s
        body = context.async_executor.run(self.step, context, resolved)
        if timeout is None:
            result = await body
        else:
            try:
                result = await asyncio.wait_for(body, timeout)
            except TimeoutError as e:  # asyncio.TimeoutError is TimeoutError on 3.11+
                raise StepTimeout(
                    f"step {type(self.step).__name__} (kind={self.step.kind or '-'}, label={self.step.label or '-'}) "
                    f"exceeded timeout_s={timeout}"
                ) from e
        return self._finalize(result, context, started_at)

    async def _aresolve_sources(self, context: ResolveContext) -> dict[str, Any]:
        """Resolve all upstream sources on the async path, one at a time.

        Sequential by design: a single shared
        per-request resource (e.g. one ``AsyncSession``) cannot run concurrent operations, and the
        ``context._resolved`` memo already collapses shared upstreams to one resolution. (Concurrency
        across independent siblings is a possible future opt-in, not the safe default.)
        """
        resolved: dict[str, Any] = {}
        for name, source in self.sources.items():
            if isinstance(source, Config):
                resolved[name] = source.value
            elif isinstance(source, Deferred):
                resolved[name] = await source.aresolve(context)
            else:
                resolved[name] = source
        return resolved

    def content_hash(self, params: dict[str, Any] | None = None) -> str | None:
        """Content hash: step identity + source hashes (memoized per params identity)."""
        if self._hash_computed and params is self._hash_params:
            return self._hash_value
        result = self._compute_content_hash(params)
        self._hash_computed = True
        self._hash_params = params
        self._hash_value = result
        return result

    def _compute_content_hash(self, params: dict[str, Any] | None = None) -> str | None:
        h = hashlib.sha256()
        for _, _, _, chunk in self._key_chunks(params):
            if chunk is None:
                return None
            h.update(chunk)
        return h.hexdigest()[:12]

    def explain_key(self, params: dict[str, Any] | None = None) -> KeyExplanation:
        """List every component of this node's cache key.

        Use it when a node recomputes and you do not know why. The components are the step's
        identity, code digest and fields, then one per input. The key is built from the same
        components, so ``explain_key(p).key == content_hash(p)``.

        Args:
            params: the run's params, as passed to ``execute``.
        """
        step = self.step
        components: list[KeyComponent] = []
        blocked_by: str | None = None
        pinned = type(step).cache_key
        if pinned is not None:
            components.append(KeyComponent.build(ComponentKind.PINNED, "", pinned, pinned.encode()))
        elif _overrides_content_hash(type(step)):
            # A subclass with its own content_hash: its parts are not visible to adag.
            own = step.content_hash()
            chunk = None if own is None else own.encode()
            components.append(KeyComponent.build(ComponentKind.IDENTITY, "", own, chunk))
            if own is None:
                blocked_by = "content_hash"
        else:
            for kind, name, value, chunk in step._key_chunks():
                components.append(KeyComponent.build(kind, name, value, chunk))
                if chunk is None and blocked_by is None:
                    blocked_by = name or kind.value
        for kind, name, value, chunk in self._key_chunks(params):
            if kind is ComponentKind.IDENTITY:
                continue  # the step's own key: already listed component by component above
            shown: Any = value
            if isinstance(value, StepResult):
                upstream: StepResult[Any] = value  # pyright: ignore[reportUnknownVariableType]
                shown = f"{type(upstream.step).__name__} {upstream.content_hash(params=params)}"
            components.append(KeyComponent.build(kind, name, shown, chunk))
            if chunk is None and blocked_by is None:
                blocked_by = name or kind.value
        label = getattr(step, "node_label", None) or None
        return KeyExplanation(
            step=type(step).__name__,
            label=label,
            key=self.content_hash(params=params),
            blocked_by=blocked_by,
            key_format=KEY_FORMAT,
            components=tuple(components),
        )

    def _key_chunks(
        self, params: dict[str, Any] | None
    ) -> Iterator[tuple[ComponentKind, str, Any, bytes | None]]:
        """Yield each component of this node's key: ``(kind, name, value, bytes)``.

        The first is the step's own key; then one per input, sorted by name. ``None`` bytes
        block the key (a missing file, an upstream with no key).
        """
        step_hash = self.step.content_hash()
        yield (
            ComponentKind.IDENTITY,
            "",
            step_hash,
            None if step_hash is None else step_hash.encode(),
        )

        for name, source in sorted(self.sources.items()):
            prefix = encode_value(name)
            if isinstance(source, StepResult):
                upstream_hash = source.content_hash(params=params)
                chunk = (
                    None if upstream_hash is None else prefix + b"UP:" + encode_value(upstream_hash)
                )
                yield ComponentKind.UPSTREAM, name, source, chunk
            elif isinstance(source, InputParam):
                if params and source.name in params:
                    param_value: Any = params[source.name]
                elif source.has_default:
                    param_value = source.default
                else:
                    param_value = None
                where = f"value of InputParam {source.name!r}"
                chunk = (
                    prefix
                    + b"PARAM:"
                    + encode_value(source.name)
                    + encode_value(param_value, where)
                )
                yield ComponentKind.PARAM, name, param_value, chunk
            elif isinstance(source, Deferred):
                # A Deferred's own content_hash (FileSource hashes file bytes) must reach the
                # key, or editing a tracked file would not invalidate anything downstream.
                deferred_hash = source.content_hash() if hasattr(source, "content_hash") else None
                chunk = (
                    prefix
                    + b"DEFERRED:"
                    + encode_value(type(source).__name__)
                    + encode_value(deferred_hash)
                )
                yield ComponentKind.DEFERRED, name, source, chunk
            else:
                where = f"input {name!r} of {type(self.step).__name__}"
                chunk = prefix + b"RAW:" + encode_value(source, where)
                yield ComponentKind.RAW, name, source, chunk

    def is_cached(self, where: Path | CacheStore, params: dict[str, Any] | None = None) -> bool:
        """Return True when a complete, valid result for this node is in ``where``.

        ``where`` is a cache directory or a ``CacheStore``. For a step that overrides
        ``save_result``, ``where`` must be a directory, and the check calls ``load_result``: an
        entry that ``load_result`` rejects is not cached, the same answer the resolver gives.
        This makes the check as costly as one load for such a step.
        """
        if self.step.caches_to_path():
            return self._try_load_path(_path_root(where, self.step), params) is not None
        return self._try_load(where, params=params) is not None

    def _result_dir(self, context: ResolveContext) -> Path | None:
        """The directory for a path result: ``cache_dir``. A ``store`` cannot hold a path.

        Raises:
            ValueError: the run has a ``store`` and no ``cache_dir``. Raised before the body runs.
        """
        if context.cache_dir is None and context.store is not None:
            raise ValueError(
                f"{type(self.step).__name__} caches to a path (it overrides save_result); "
                "a CacheStore holds only text. Pass cache_dir= as well."
            )
        return context.cache_dir

    def _try_load_path(self, cache_dir: Path | None, params: dict[str, Any] | None) -> Out | None:
        """Load a path result with ``load_result``; None on a miss or an invalid entry."""
        key = self.content_hash(params=params)
        if cache_dir is None or key is None:
            return None
        path = cache_dir.joinpath(key)
        if not cache_dir.joinpath(key + ".done").exists() or not path.exists():
            return None
        try:
            return self.step.load_result(path)
        except Exception as exc:
            emit("step.cache.invalid_entry", step=type(self.step).__name__, key=key, error=str(exc))
            return None

    def _save_path(self, result: Out, cache_dir: Path, params: dict[str, Any] | None) -> Out:
        """Save with ``save_result`` to a temp path, rename it into place, write the marker, and
        return ``load_result`` of the final path.

        Returning the reloaded value means a fresh run and a cached run hand downstream the same
        thing, and a result that refers to its own location gets the final path.
        """
        key = self.content_hash(params=params)
        if key is None:
            return result
        cache_dir.mkdir(parents=True, exist_ok=True)
        final = cache_dir.joinpath(key)
        tmp = cache_dir.joinpath(f"{key}.tmp-{os.getpid()}-{uuid4().hex[:8]}")
        try:
            self.step.save_result(result, tmp)
            cache_dir.joinpath(key + ".done").unlink(missing_ok=True)
            _remove_path(final)
            tmp.rename(final)
        finally:
            _remove_path(tmp)
        atomic_write_text(cache_dir.joinpath(key + ".done"), datetime.now(UTC).isoformat())
        reloaded = self.step.load_result(final)
        step_cls = type(self.step)
        if step_cls not in _ROUND_TRIP_WARNED and _round_trip_differs(result, reloaded):
            _ROUND_TRIP_WARNED.add(step_cls)
            logger.warning(
                "step.load_result_differs",
                step=step_cls.__name__,
                hint="downstream gets load_result's value, also on the first run; "
                "load_result must rebuild everything run put in the result",
            )
        return reloaded

    def _try_load(
        self, where: Path | CacheStore, params: dict[str, Any] | None = None
    ) -> Out | None:
        """Read and validate the cached result in one pass; None on any miss.

        A missing ``.done`` marker, an empty value, or a value the result model rejects is a
        miss, never a crash. Invalid entries are reported so a broken cache surfaces.
        """
        result_type = self.step.result_model()
        key = self.content_hash(params=params)
        if result_type is None or key is None:
            return None
        store = _as_store(where)
        if not store.exists(key + ".done"):
            return None
        text = store.get(key)
        if text is None:
            return None
        try:
            return result_type.model_validate_json(text)  # pyright: ignore[reportReturnType]
        except Exception as exc:
            emit("step.cache.invalid_entry", step=type(self.step).__name__, key=key, error=str(exc))
            return None

    def _save(
        self, result: Out, where: Path | CacheStore, params: dict[str, Any] | None = None
    ) -> None:
        key = self.content_hash(params=params)
        if key is None:
            return
        if not isinstance(result, BaseModel):
            raise TypeError(
                f"Cannot cache {type(result).__name__}: result_type is set but result is not a BaseModel"
            )
        store = _as_store(where)
        new_text = result.model_dump_json(indent=2)
        # Invariant: same content_hash -> same bytes. Overwriting an existing hash with
        # different bytes means the cache key is missing an input; log loudly (the message
        # below has the likely causes) rather than silently ship stale results.
        old_text = store.get(key)
        if old_text is not None and old_text != new_text:
            logger.error(
                "cached_step.invariant_violation",
                step=type(self.step).__name__,
                hash=key,
                message=(
                    "Same content_hash produced different bytes. The "
                    "step's cache key is missing an input — likely a "
                    "side-channel file (model weights, embeddings) or "
                    "a sub-batch whose item keys do not encode all "
                    "inputs. Overwriting cache with new bytes."
                ),
            )
        store.set(key, new_text)
        store.set(key + ".done", datetime.now(UTC).isoformat())

    def load(self, where: Path | CacheStore, params: dict[str, Any] | None = None) -> Out:
        """Load the cached result from a cache directory or a ``CacheStore``.

        Raises:
            ValueError: no result is stored for this node.
        """
        if self.step.caches_to_path():
            loaded = self._try_load_path(_path_root(where, self.step), params)
            if loaded is None:
                raise ValueError(f"Output for '{type(self.step).__name__}' not found.")
            return loaded
        key = self.content_hash(params=params)
        result_type = self.step.result_model()
        text = _as_store(where).get(key) if key is not None else None
        if text is None or result_type is None:
            raise ValueError(f"Output for '{type(self.step).__name__}' not found.")
        return result_type.model_validate_json(text)  # pyright: ignore[reportReturnType]

    def __repr__(self) -> str:
        return f"StepResult({type(self.step).__name__})"

resolve

resolve(context: ResolveContext) -> Out

Resolve upstream sources, run the step, cache if applicable.

Memoizes by id(self) so multiple downstream consumers of the SAME StepResult share one computed result within a Pipeline.execute. This is critical for non-cached steps returning non-serializable objects (e.g. model handles), where consumers must see the same instance, not two re-executions. The cache/memo/observer skeleton lives in _check_cache/_finalize (shared with aresolve); only the two lines that genuinely differ (upstream resolution and the step body) stay here, run inline.

In-flight dedup makes that guarantee hold under CONCURRENCY too: the memo only collapses resolutions that do not overlap in time, so N branch workers reaching one shared upstream at once would each miss the memo and run the body N times. Here the first thread to miss the memo claims ownership (an Event in context._inflight); concurrent threads find that Event, wait on it, then loop: a success left the value in the memo (returned), a failure removed the marker so they re-claim and resolve it themselves. Uncontended (single-threaded) there is never a waiter, so behaviour and cost are unchanged bar creating one Event that is set immediately.

Source code in src/adag/step.py

def resolve(self, context: ResolveContext) -> Out:
    """Resolve upstream sources, run the step, cache if applicable.

    Memoizes by ``id(self)`` so multiple downstream consumers of the SAME ``StepResult``
    share one computed result within a ``Pipeline.execute``. This is critical for non-cached
    steps returning non-serializable objects (e.g. model handles), where consumers must
    see the same instance, not two re-executions. The cache/memo/observer skeleton lives
    in ``_check_cache``/``_finalize`` (shared with ``aresolve``); only the two lines that
    genuinely differ (upstream resolution and the step body) stay here, run inline.

    In-flight dedup makes that guarantee hold under CONCURRENCY too: the memo only
    collapses resolutions that do not overlap in time, so N branch workers reaching one
    shared upstream at once would each miss the memo and run the body N times. Here the
    first thread to miss the memo claims ownership (an ``Event`` in ``context._inflight``);
    concurrent threads find that Event, wait on it, then loop: a success left the value in
    the memo (returned), a failure removed the marker so they re-claim and resolve it
    themselves. Uncontended (single-threaded) there is never a waiter, so behaviour and
    cost are unchanged bar creating one Event that is set immediately.
    """
    while True:
        with context._lock:
            memo = self._memo_get(context)
            if not isinstance(memo, _Miss):
                return memo
            in_flight = context._inflight.get(id(self))
            if in_flight is None:
                own_event = Event()
                context._inflight[id(self)] = own_event
                break
        # Another thread owns this exact node's resolution. Wait for it, then loop.
        in_flight.wait()
    try:
        cached = self._check_cache(context)
        if not isinstance(cached, _Miss):
            return cached
        resolved = self._merge_step_defaults(self._resolve_sources(context))
        started_at = time.perf_counter()
        result = context.executor.run(self.step, context, resolved)
        return self._finalize(result, context, started_at)
    finally:
        with context._lock:
            context._inflight.pop(id(self), None)
        own_event.set()

aresolve async

aresolve(context: ResolveContext) -> Out

Resolve this node on the event loop and return its result. Public API.

Use it to schedule async steps yourself, for example one asyncio.gather per topological level inside an async request handler. Pass the same ResolveContext to every call: it memoizes each node, so a shared upstream runs once and a later call reuses earlier results. Upstream nodes are awaited, async step bodies run on the loop, and sync bodies run unchanged. A cancelled task starts no further step.

Gathered calls that reach the same node at the same time run its body once: the first task claims the node and the others wait for it. If that body raises, a waiting task resolves the node itself.

Source code in src/adag/step.py

async def aresolve(self, context: ResolveContext) -> Out:
    """Resolve this node on the event loop and return its result. Public API.

    Use it to schedule async steps yourself, for example one ``asyncio.gather`` per
    topological level inside an async request handler. Pass the same ``ResolveContext`` to
    every call: it memoizes each node, so a shared upstream runs once and a later call reuses
    earlier results. Upstream nodes are awaited, async step bodies run on the loop, and sync
    bodies run unchanged. A cancelled task starts no further step.

    Gathered calls that reach the same node at the same time run its body once: the first
    task claims the node and the others wait for it. If that body raises, a waiting task
    resolves the node itself.
    """
    while True:
        with context._lock:
            memo = self._memo_get(context)
            if not isinstance(memo, _Miss):
                return memo
            in_flight = context._ainflight.get(id(self))
            if in_flight is None:
                own = asyncio.get_running_loop().create_future()
                context._ainflight[id(self)] = own
                break
        # Another task owns this node. Wait without taking its exception, then loop: a
        # success left the value in the memo, a failure removed the marker.
        await asyncio.wait([in_flight])
    try:
        return await self._aresolve_owned(context)
    finally:
        with context._lock:
            context._ainflight.pop(id(self), None)
        if not own.done():
            own.set_result(None)

content_hash

content_hash(
    params: dict[str, Any] | None = None,
) -> str | None

Content hash: step identity + source hashes (memoized per params identity).

Source code in src/adag/step.py

def content_hash(self, params: dict[str, Any] | None = None) -> str | None:
    """Content hash: step identity + source hashes (memoized per params identity)."""
    if self._hash_computed and params is self._hash_params:
        return self._hash_value
    result = self._compute_content_hash(params)
    self._hash_computed = True
    self._hash_params = params
    self._hash_value = result
    return result

explain_key

explain_key(
    params: dict[str, Any] | None = None,
) -> KeyExplanation

List every component of this node's cache key.

Use it when a node recomputes and you do not know why. The components are the step's identity, code digest and fields, then one per input. The key is built from the same components, so explain_key(p).key == content_hash(p).

PARAMETER	DESCRIPTION
params	the run's params, as passed to execute. TYPE: dict[str, Any] | None DEFAULT: None

Source code in src/adag/step.py

def explain_key(self, params: dict[str, Any] | None = None) -> KeyExplanation:
    """List every component of this node's cache key.

    Use it when a node recomputes and you do not know why. The components are the step's
    identity, code digest and fields, then one per input. The key is built from the same
    components, so ``explain_key(p).key == content_hash(p)``.

    Args:
        params: the run's params, as passed to ``execute``.
    """
    step = self.step
    components: list[KeyComponent] = []
    blocked_by: str | None = None
    pinned = type(step).cache_key
    if pinned is not None:
        components.append(KeyComponent.build(ComponentKind.PINNED, "", pinned, pinned.encode()))
    elif _overrides_content_hash(type(step)):
        # A subclass with its own content_hash: its parts are not visible to adag.
        own = step.content_hash()
        chunk = None if own is None else own.encode()
        components.append(KeyComponent.build(ComponentKind.IDENTITY, "", own, chunk))
        if own is None:
            blocked_by = "content_hash"
    else:
        for kind, name, value, chunk in step._key_chunks():
            components.append(KeyComponent.build(kind, name, value, chunk))
            if chunk is None and blocked_by is None:
                blocked_by = name or kind.value
    for kind, name, value, chunk in self._key_chunks(params):
        if kind is ComponentKind.IDENTITY:
            continue  # the step's own key: already listed component by component above
        shown: Any = value
        if isinstance(value, StepResult):
            upstream: StepResult[Any] = value  # pyright: ignore[reportUnknownVariableType]
            shown = f"{type(upstream.step).__name__} {upstream.content_hash(params=params)}"
        components.append(KeyComponent.build(kind, name, shown, chunk))
        if chunk is None and blocked_by is None:
            blocked_by = name or kind.value
    label = getattr(step, "node_label", None) or None
    return KeyExplanation(
        step=type(step).__name__,
        label=label,
        key=self.content_hash(params=params),
        blocked_by=blocked_by,
        key_format=KEY_FORMAT,
        components=tuple(components),
    )

is_cached

is_cached(
    where: Path | CacheStore,
    params: dict[str, Any] | None = None,
) -> bool

Return True when a complete, valid result for this node is in where.

where is a cache directory or a CacheStore. For a step that overrides save_result, where must be a directory, and the check calls load_result: an entry that load_result rejects is not cached, the same answer the resolver gives. This makes the check as costly as one load for such a step.

Source code in src/adag/step.py

def is_cached(self, where: Path | CacheStore, params: dict[str, Any] | None = None) -> bool:
    """Return True when a complete, valid result for this node is in ``where``.

    ``where`` is a cache directory or a ``CacheStore``. For a step that overrides
    ``save_result``, ``where`` must be a directory, and the check calls ``load_result``: an
    entry that ``load_result`` rejects is not cached, the same answer the resolver gives.
    This makes the check as costly as one load for such a step.
    """
    if self.step.caches_to_path():
        return self._try_load_path(_path_root(where, self.step), params) is not None
    return self._try_load(where, params=params) is not None

load

load(
    where: Path | CacheStore,
    params: dict[str, Any] | None = None,
) -> Out

Load the cached result from a cache directory or a CacheStore.

RAISES	DESCRIPTION
ValueError	no result is stored for this node.

Source code in src/adag/step.py

def load(self, where: Path | CacheStore, params: dict[str, Any] | None = None) -> Out:
    """Load the cached result from a cache directory or a ``CacheStore``.

    Raises:
        ValueError: no result is stored for this node.
    """
    if self.step.caches_to_path():
        loaded = self._try_load_path(_path_root(where, self.step), params)
        if loaded is None:
            raise ValueError(f"Output for '{type(self.step).__name__}' not found.")
        return loaded
    key = self.content_hash(params=params)
    result_type = self.step.result_model()
    text = _as_store(where).get(key) if key is not None else None
    if text is None or result_type is None:
        raise ValueError(f"Output for '{type(self.step).__name__}' not found.")
    return result_type.model_validate_json(text)  # pyright: ignore[reportReturnType]

StepTimeout

Bases: TimeoutError

Raised when a step's async body exceeds its timeout_s budget. Subclasses the builtin TimeoutError so callers can catch either.

Source code in src/adag/step.py

class StepTimeout(TimeoutError):
    """Raised when a step's async body exceeds its ``timeout_s`` budget. Subclasses the builtin
    ``TimeoutError`` so callers can catch either.
    """

SVG

Wrapper around an SVG string that Jupyter renders as an image.

Source code in src/adag/visualization.py

class SVG:
    """Wrapper around an SVG string that Jupyter renders as an image."""

    def __init__(self, source: str) -> None:
        self.source = source

    def _repr_svg_(self) -> str:
        return self.source

    def __str__(self) -> str:
        return self.source

Edge dataclass

A directed edge in the walked graph.

Source code in src/adag/visualization.py

@dataclass(frozen=True)
class Edge:
    """A directed edge in the walked graph."""

    source: str
    target: str
    label: str

Node dataclass

A node in the walked graph.

name is the node's unique KEY (edges reference it, layout positions by it). title, when set, is the human display label; empty → the renderer derives one from name. A Step may expose a per-INSTANCE label via a node_label attribute (walk() then uses it as both name and title; the name gets a suffix on collision) so two instances of the SAME Step class render as two distinct boxes instead of collapsing into one.

Source code in src/adag/visualization.py

@dataclass(frozen=True)
class Node:
    """A node in the walked graph.

    `name` is the node's unique KEY (edges reference it, layout positions by it). `title`, when set, is the
    human display label; empty → the renderer derives one from `name`. A Step may expose a per-INSTANCE label
    via a `node_label` attribute (walk() then uses it as both name and title; the name gets a suffix on collision) so
    two instances of the SAME Step class render as two distinct boxes instead of collapsing into one.
    """

    name: str
    step: Step[Any] | None
    kind: str
    title: str = ""
    result: StepResult[Any] | None = None  # the graph node; its cache key covers the inputs

NodeMetadata dataclass

Pre-extracted display info for a graph node, decoupled from Step introspection.

Source code in src/adag/visualization.py

@dataclass(frozen=True)
class NodeMetadata:
    """Pre-extracted display info for a graph node, decoupled from Step introspection."""

    name: str
    kind: str
    title: str
    description: str = ""
    params: tuple[tuple[str, str], ...] = ()
    state: str = "PENDING"

SvgOptions dataclass

What the renderer emits.

include_styles=False leaves the SVG unpainted apart from its classes, for a host that supplies the whole stylesheet. show_params=False drops the per-step parameter lines, which also shrinks the boxes: the layout is computed from the same node data that gets drawn.

Source code in src/adag/visualization.py

@dataclass(frozen=True)
class SvgOptions:
    """What the renderer emits.

    ``include_styles=False`` leaves the SVG unpainted apart from its classes,
    for a host that supplies the whole stylesheet. ``show_params=False`` drops
    the per-step parameter lines, which also shrinks the boxes: the layout is
    computed from the same node data that gets drawn.
    """

    theme: str = "light"
    include_styles: bool = True
    include_background: bool = True
    show_params: bool = True

    def stylesheet(self) -> str:
        return _STYLESHEETS.get(self.theme, _STYLESHEET_LIGHT)

current_observer

current_observer() -> PipelineObserver

The observer bound for the in-flight step, or a NullObserver outside any run.

Set for the duration of a step body by :func:observing (which Step.execute enters), so nested asyncio tasks and worker threads can emit hooks without threading an observer argument through every signature. Propagates across asyncio.to_thread and asyncio tasks automatically; a raw ThreadPoolExecutor submit must run inside a copied context (contextvars.copy_context().run) to inherit it.

Source code in src/adag/deferred.py

def current_observer() -> PipelineObserver:
    """The observer bound for the in-flight step, or a NullObserver outside any run.

    Set for the duration of a step body by :func:`observing` (which ``Step.execute`` enters), so nested
    asyncio tasks and worker threads can emit hooks
    without threading an observer argument through every signature. Propagates across ``asyncio.to_thread``
    and asyncio tasks automatically; a raw ``ThreadPoolExecutor`` submit must run inside a copied
    context (``contextvars.copy_context().run``) to inherit it.
    """
    return _CURRENT_OBSERVER.get()

observing

observing(observer: PipelineObserver) -> Generator[None]

Bind observer as the ambient :func:current_observer for the duration of the block.

Source code in src/adag/deferred.py

@contextlib.contextmanager
def observing(observer: PipelineObserver) -> Generator[None]:
    """Bind ``observer`` as the ambient :func:`current_observer` for the duration of the block."""
    token = _CURRENT_OBSERVER.set(observer)
    try:
        yield
    finally:
        _CURRENT_OBSERVER.reset(token)

aresolve_all async

aresolve_all(
    targets: Sequence[StepResult[Any]],
    context: ResolveContext,
    *,
    max_concurrency: int | None = None,
) -> list[Any]

Resolve targets level by level on the running loop; results in targets order.

Use it inside your own event loop (a request handler, a script) instead of sorting the graph and gathering aresolve per level yourself. context holds the params, observer, cache and the memo shared by every node.

• Each level's nodes run together with asyncio.gather; the next level starts when this one is done. A node with a saved result is not expanded: its upstream nodes do not run.
• Nodes in one level with the same cache key run once.
• max_concurrency bounds how many step bodies run at once. None runs a whole level.
• The first failure cancels the rest of its level and is raised as it is (no ExceptionGroup).
• Sync step bodies and cache reads run on the loop and block it while they run.
• FanOutStep branches are resolved inside the step, so max_concurrency does not bound them.

PARAMETER	DESCRIPTION
targets	the nodes to resolve. TYPE: Sequence[StepResult[Any]]
context	shared by every node; build one per run. TYPE: ResolveContext
max_concurrency	the most step bodies running at the same time. TYPE: int | None DEFAULT: None

Source code in src/adag/pipeline.py

async def aresolve_all(
    targets: Sequence[StepResult[Any]],
    context: ResolveContext,
    *,
    max_concurrency: int | None = None,
) -> list[Any]:
    """Resolve ``targets`` level by level on the running loop; results in ``targets`` order.

    Use it inside your own event loop (a request handler, a script) instead of sorting the graph
    and gathering ``aresolve`` per level yourself. ``context`` holds the params, observer, cache
    and the memo shared by every node.

    - Each level's nodes run together with ``asyncio.gather``; the next level starts when this
      one is done. A node with a saved result is not expanded: its upstream nodes do not run.
    - Nodes in one level with the same cache key run once.
    - ``max_concurrency`` bounds how many step bodies run at once. ``None`` runs a whole level.
    - The first failure cancels the rest of its level and is raised as it is (no
      ``ExceptionGroup``).
    - Sync step bodies and cache reads run on the loop and block it while they run.
    - ``FanOutStep`` branches are resolved inside the step, so ``max_concurrency`` does not
      bound them.

    Args:
        targets: the nodes to resolve.
        context: shared by every node; build one per run.
        max_concurrency: the most step bodies running at the same time.
    """
    semaphore = asyncio.Semaphore(max_concurrency) if max_concurrency else None

    async def run(node: StepResult[Any]) -> Any:
        if semaphore is None:
            return await node.aresolve(context)
        async with semaphore:
            return await node.aresolve(context)

    with observing(context.observer):
        for level in _node_levels(targets, lambda n: _has_cache_entry(n, context)):
            leaders: list[StepResult[Any]] = []
            followers: dict[int, list[StepResult[Any]]] = defaultdict(list)
            leader_by_key: dict[str, StepResult[Any]] = {}
            for node in level:
                key = _key_or_none(node, context)
                leader = leader_by_key.get(key) if key is not None else None
                if leader is None:
                    leaders.append(node)
                    if key is not None:
                        leader_by_key[key] = node
                else:
                    followers[id(leader)].append(node)
            tasks = [asyncio.create_task(run(node)) for node in leaders]
            try:
                values = await _gather_first_failure(tasks)
            finally:
                for task in tasks:
                    task.cancel()
            for leader, value in zip(leaders, values, strict=True):
                with context._lock:
                    for follower in followers[id(leader)]:
                        context._resolved[id(follower)] = (follower, value)
        return [await target.aresolve(context) for target in targets]

compute_state

compute_state(node: Node, cache_dir: Path | None) -> str

Return the state a node is drawn with: CACHED, PENDING, BLOCKED or SOURCE.

SOURCE is an InputParam or a file. BLOCKED is a step with no hash, for example a FileSource field whose file is missing. Without cache_dir every step is PENDING. The key is computed with InputParam defaults, not with the values of a run.

Source code in src/adag/visualization.py

def compute_state(node: Node, cache_dir: Path | None) -> str:
    """Return the state a node is drawn with: ``CACHED``, ``PENDING``, ``BLOCKED`` or ``SOURCE``.

    ``SOURCE`` is an ``InputParam`` or a file. ``BLOCKED`` is a step with no hash, for example a
    ``FileSource`` field whose file is missing. Without ``cache_dir`` every step is ``PENDING``.
    The key is computed with ``InputParam`` defaults, not with the values of a run.
    """
    if node.step is None:
        return "SOURCE"
    if cache_dir is None:
        return "PENDING"
    h = node.step.content_hash() if hasattr(node.step, "content_hash") else None
    if h is None:
        return "BLOCKED"
    if node.result is not None:
        # The pipeline caches under the node key (step and inputs), not the step hash.
        return "CACHED" if node.result.is_cached(cache_dir) else "PENDING"
    is_cached_attr: object = getattr(node.step, "is_cached", None)
    if is_cached_attr is not None:
        cached = is_cached_attr(cache_dir) if callable(is_cached_attr) else is_cached_attr
        return "CACHED" if cached else "PENDING"
    return "PENDING"

dag

dag(
    output: Deferred[Any],
    *,
    cache_dir: Path | None = None,
    path: str | Path | None = None,
    prefix: str = "",
    options: SvgOptions = _DEFAULT_SVG_OPTIONS,
) -> SVG

Render the DAG as SVG.

PARAMETER	DESCRIPTION
output	Any Deferred: walks backward to discover the full graph. TYPE: Deferred[Any]
cache_dir	Cache directory (for cached/pending state coloring). TYPE: Path | None DEFAULT: None
path	Save SVG to this path. TYPE: str | Path | None DEFAULT: None
prefix	Text removed from the start of step names in titles. TYPE: str DEFAULT: ''
options	Theme, and whether to embed the stylesheet, paint a background or draw the per-step parameters. TYPE: SvgOptions DEFAULT: _DEFAULT_SVG_OPTIONS

Source code in src/adag/visualization.py

def dag(
    output: Deferred[Any],
    *,
    cache_dir: Path | None = None,
    path: str | Path | None = None,
    prefix: str = "",
    options: SvgOptions = _DEFAULT_SVG_OPTIONS,
) -> SVG:
    """Render the DAG as SVG.

    Args:
        output: Any Deferred: walks backward to discover the full graph.
        cache_dir: Cache directory (for cached/pending state coloring).
        path: Save SVG to this path.
        prefix: Text removed from the start of step names in titles.
        options: Theme, and whether to embed the stylesheet, paint a background
            or draw the per-step parameters.
    """
    graph_nodes, graph_edges = walk(output)
    return SVG(
        _render_svg(
            output,
            graph_nodes,
            graph_edges,
            cache_dir=cache_dir,
            prefix=prefix,
            path=path,
            options=options,
        )
    )

node_metadata

node_metadata(
    nodes: list[Node],
    *,
    cache_dir: Path | None = None,
    prefix: str = "",
) -> list[NodeMetadata]

Return the display data for each node from walk, in the same order.

Use it to draw a pipeline your own way: walk the graph, change the NodeMetadata you want (dataclasses.replace a title, a description or the params), then pass the list to render_graph_svg. dag does the same with no changes.

PARAMETER	DESCRIPTION
nodes	the nodes returned by walk. TYPE: list[Node]
cache_dir	colour each step by its cache state, as dag(..., cache_dir=) does. TYPE: Path | None DEFAULT: None
prefix	strip this from step names in the titles. TYPE: str DEFAULT: ''

Source code in src/adag/visualization.py

def node_metadata(
    nodes: list[Node], *, cache_dir: Path | None = None, prefix: str = ""
) -> list[NodeMetadata]:
    """Return the display data for each node from ``walk``, in the same order.

    Use it to draw a pipeline your own way: walk the graph, change the ``NodeMetadata`` you want
    (``dataclasses.replace`` a title, a description or the params), then pass the list to
    ``render_graph_svg``. ``dag`` does the same with no changes.

    Args:
        nodes: the nodes returned by ``walk``.
        cache_dir: colour each step by its cache state, as ``dag(..., cache_dir=)`` does.
        prefix: strip this from step names in the titles.
    """
    by_name = _walk_to_metadata(nodes, cache_dir=cache_dir, prefix=prefix)
    return [by_name[node.name] for node in nodes]

render_graph_svg

render_graph_svg(
    nodes: list[NodeMetadata],
    edges: list[tuple[str, str]],
    *,
    output_label: str = "",
    final_step_name: str | None = None,
    options: SvgOptions = _DEFAULT_SVG_OPTIONS,
) -> str

Return the SVG for explicit node metadata and (source, target) edge pairs.

Use it to draw a graph that is not a live pipeline, for example one built from your own records. dag(output) and render_pipeline_svg(pipeline_dir) call the same layout.

Source code in src/adag/visualization.py

def render_graph_svg(
    nodes: list[NodeMetadata],
    edges: list[tuple[str, str]],
    *,
    output_label: str = "",
    final_step_name: str | None = None,
    options: SvgOptions = _DEFAULT_SVG_OPTIONS,
) -> str:
    """Return the SVG for explicit node metadata and ``(source, target)`` edge pairs.

    Use it to draw a graph that is not a live pipeline, for example one built from your own
    records. ``dag(output)`` and ``render_pipeline_svg(pipeline_dir)`` call the same layout.
    """
    # Strip params before layout, not at paint time: the box sizes are computed
    # from the same metadata, so dropping them later would leave empty space.
    if not options.show_params:
        nodes = [replace(n, params=()) for n in nodes]
    nodes_dict = {n.name: n for n in nodes}
    layout_nodes, edge_pairs = _compute_layout(
        nodes_dict, edges, output_label=output_label, final_step_name=final_step_name
    )
    return _render_layout(layout_nodes, edge_pairs, options)

render_pipeline_svg

render_pipeline_svg(
    pipeline_dir: str | Path,
    *,
    options: SvgOptions = _DEFAULT_SVG_OPTIONS,
) -> str | None

Render the pipeline at pipeline_dir as SVG from its on-disk manifest.

Reads pipeline.json and each step's step.json, builds the graph, and produces an SVG string. Returns None when there is no manifest or the manifest declares no steps.

Source code in src/adag/visualization.py

def render_pipeline_svg(
    pipeline_dir: str | Path, *, options: SvgOptions = _DEFAULT_SVG_OPTIONS
) -> str | None:
    """Render the pipeline at ``pipeline_dir`` as SVG from its on-disk manifest.

    Reads ``pipeline.json`` and each step's ``step.json``, builds the graph,
    and produces an SVG string. Returns ``None`` when there is no manifest
    or the manifest declares no steps.
    """
    base = str(pipeline_dir)
    pipeline_path = os.path.join(base, "pipeline.json")
    if not os.path.exists(pipeline_path):
        return None
    with open(pipeline_path) as f:
        pipeline_data = json.load(f)

    step_dirs: list[str] = pipeline_data.get("steps", [])
    if not step_dirs:
        return None

    nodes: list[NodeMetadata] = []
    for step_dir in step_dirs:
        step_json_path = os.path.join(base, step_dir, "step.json")
        spec: dict[str, Any] = {}
        kind = ""
        if os.path.exists(step_json_path):
            with open(step_json_path) as f:
                step_data = json.load(f)
            spec = step_data.get("spec", {})
            kind = step_data.get("kind", "")

        nodes.append(
            NodeMetadata(
                name=step_dir,
                kind="step",
                title=_humanize_step_name(step_dir),
                description=kind,
                params=_scalar_params_from_spec(spec),
            )
        )

    edges: list[tuple[str, str]] = []
    for edge in pipeline_data.get("edges", []):
        source = edge.get("source", "")
        target = edge.get("target", "")
        if source.startswith("parameter:"):
            continue
        src_node = source.split(".", 1)[0]
        tgt_node = target.split(".", 1)[0]
        if src_node and tgt_node:
            edges.append((src_node, tgt_node))

    final_step_name = pipeline_data.get("output") or step_dirs[-1]
    return render_graph_svg(
        nodes, edges, output_label="", final_step_name=final_step_name, options=options
    )

walk

walk(
    output: Deferred[Any],
) -> tuple[list[Node], list[Edge]]

Walk any Deferred backward and return the full dependency graph.

Step nodes are identified by instance (id(step)), not class name. A shared upstream (the same instance reached on two paths) is one node. Two distinct instances of the same class are two nodes. Each node's label is its node_label attribute when present, else the class name. The key gets a suffix (Name#2) only when two instances would otherwise collide.

Source code in src/adag/visualization.py

def walk(output: Deferred[Any]) -> tuple[list[Node], list[Edge]]:
    """Walk any Deferred backward and return the full dependency graph.

    Step nodes are identified by instance (``id(step)``), not class name. A shared upstream (the
    same instance reached on two paths) is one node. Two distinct instances of the same class are
    two nodes. Each node's label is its ``node_label`` attribute when present, else the class name.
    The key gets a suffix (``Name#2``) only when two instances would otherwise collide.
    """
    nodes: dict[str, Node] = {}
    edges: list[Edge] = []
    placed: dict[int, str] = {}  # id(step) → the unique node name assigned to that instance

    def _visit(deferred: Any) -> str:
        leaf = _node_name_and_kind(deferred)
        if leaf is not None:
            name, kind = leaf
            if name not in nodes:
                nodes[name] = Node(name=name, step=None, kind=kind)
            return name

        # Capture .step while deferred is still Any (before isinstance narrows it).
        step: Any = deferred.step if hasattr(deferred, "step") else None

        if isinstance(deferred, StepResult):
            if id(step) in placed:  # same instance reached again (shared upstream) → the same node
                return placed[id(step)]
            label = getattr(step, "node_label", "") or ""
            base = label or type(step).__name__
            name = base
            suffix = 2
            while (
                name in nodes
            ):  # a DIFFERENT instance wants the same base name → disambiguate the key
                name = f"{base}#{suffix}"
                suffix += 1
            placed[id(step)] = name
            nodes[name] = Node(
                name=name,
                step=step,
                kind="step",
                title=label,
                result=cast("StepResult[Any]", deferred),
            )
            for input_name, source in deferred.sources.items():
                if isinstance(source, Deferred):
                    source_name = _visit(source)
                    if source_name:
                        edges.append(Edge(source=source_name, target=name, label=input_name))
            return name

        return ""

    _visit(output)
    return list(nodes.values()), edges

__getattr__

__getattr__(name: str) -> object

Resolve deprecated top-level names on first use, so import adag does not warn.

Source code in src/adag/__init__.py

def __getattr__(name: str) -> object:
    """Resolve deprecated top-level names on first use, so ``import adag`` does not warn."""
    if name in _DEPRECATED:
        import importlib

        from adag.deprecation import warn_deprecated

        module, attr, replacement = _DEPRECATED[name]
        warn_deprecated(f"adag.{name}", replacement)
        return getattr(importlib.import_module(module), attr)
    raise AttributeError(f"module 'adag' has no attribute {name!r}")

Dispatch

adag.dispatch

Branch dispatchers: resolve fan-out branches on workers, rejoin through the cache.

A BranchDispatcher (see deferred.py) decides where the branches of a FanOutStep run. The default InlineDispatcher runs them in-process.

RemoteDispatcher runs each branch as a job on a JobBackend. The worker runs adag resolve-branch, which re-imports the pipeline, re-derives the branch, resolves it and writes the result to the shared cache under the branch content hash. The dispatcher then loads the result by that hash. Only the BranchJob crosses to the worker.

SubprocessDispatcher is RemoteDispatcher with LocalJobBackend: one local process per branch.

EXIT_BAD_BRANCH module-attribute

EXIT_BAD_BRANCH = 2

Worker exit code for a bad branch reference or a hash mismatch. Never retried.

BranchExecutionError

Bases: RuntimeError

Deprecated: adag no longer raises this. Branch failures arrive as FanOutBranchError.

Kept so except BranchExecutionError and custom dispatchers that raise it still import. Removed in 1.0.

Source code in src/adag/dispatch.py

class BranchExecutionError(RuntimeError):
    """Deprecated: adag no longer raises this. Branch failures arrive as ``FanOutBranchError``.

    Kept so ``except BranchExecutionError`` and custom dispatchers that raise it still import.
    Removed in 1.0.
    """

    def __init__(self, key: str, detail: str) -> None:
        self.key = key
        super().__init__(f"branch {key!r} failed: {detail}")

BranchPolicy dataclass

How hard to try a branch and how patiently to wait for it.

timeout_seconds counts from when the job is RUNNING. queue_timeout_seconds counts from submit while the job is PENDING; a batch queue can hold a job for hours. result_grace_seconds is how long to wait for the result to appear after a job SUCCEEDED, because a network filesystem can show a new file late.

Source code in src/adag/dispatch.py

@dataclass(frozen=True)
class BranchPolicy:
    """How hard to try a branch and how patiently to wait for it.

    ``timeout_seconds`` counts from when the job is RUNNING. ``queue_timeout_seconds`` counts
    from submit while the job is PENDING; a batch queue can hold a job for hours.
    ``result_grace_seconds`` is how long to wait for the result to appear after a job
    SUCCEEDED, because a network filesystem can show a new file late.
    """

    max_retries: int = 0
    timeout_seconds: float | None = None
    poll_interval_seconds: float = 5.0
    retry_delay_seconds: float = 0.0
    result_grace_seconds: float = 5.0
    # Added in 0.2.11. Kept after the first-release fields so BranchPolicy(1, 60, 2.0) still means
    # (max_retries, timeout_seconds, poll_interval_seconds).
    queue_timeout_seconds: float | None = None

BranchReferenceError

Bases: LookupError

The worker cannot find the branch, or derives a different hash than the main process.

Source code in src/adag/dispatch.py

class BranchReferenceError(LookupError):
    """The worker cannot find the branch, or derives a different hash than the main process."""

Resources dataclass

What one branch job asks the scheduler for. None means the scheduler's default.

Source code in src/adag/dispatch.py

@dataclass(frozen=True)
class Resources:
    """What one branch job asks the scheduler for. ``None`` means the scheduler's default."""

    cpus: float | None = None
    memory_mb: int | None = None
    gpus: int = 0

BranchJob dataclass

One attempt at one fan-out branch, as a backend submits it.

A backend runs command() in an environment that has the pipeline code and can reach cache_dir, or the store at store_url. Success means the result is in the cache under branch.branch_hash. The job's exit code alone is not trusted.

Source code in src/adag/dispatch.py

@dataclass(frozen=True)
class BranchJob:
    """One attempt at one fan-out branch, as a backend submits it.

    A backend runs ``command()`` in an environment that has the pipeline code and can reach
    ``cache_dir``, or the store at ``store_url``. Success means the result is in the cache
    under ``branch.branch_hash``. The job's exit code alone is not trusted.
    """

    branch: BranchRef
    params_json: str
    cache_dir: str | None
    attempt: int = 1
    resources: Resources = Resources()
    store_url: str | None = None

    def command(self, prefix: Sequence[str] = ("adag",)) -> tuple[str, ...]:
        """The worker command line. ``prefix`` is how to start adag (``python -m adag``)."""
        return (
            *prefix,
            "resolve-branch",
            self.branch.pipeline_module,
            "--fan-out",
            self.branch.fan_out_var,
            "--key",
            self.branch.key,
            *(("--cache-dir", self.cache_dir) if self.cache_dir is not None else ()),
            *(("--store-url", self.store_url) if self.store_url is not None else ()),
            "--params-json",
            self.params_json,
            "--expect-hash",
            self.branch.branch_hash,
        )

    def name(self) -> str:
        """A job name unique per branch and attempt, for scheduler consoles."""
        return f"adag-{self.branch.branch_hash}-{self.attempt}"

command

command(
    prefix: Sequence[str] = ("adag",),
) -> tuple[str, ...]

The worker command line. prefix is how to start adag (python -m adag).

Source code in src/adag/dispatch.py

def command(self, prefix: Sequence[str] = ("adag",)) -> tuple[str, ...]:
    """The worker command line. ``prefix`` is how to start adag (``python -m adag``)."""
    return (
        *prefix,
        "resolve-branch",
        self.branch.pipeline_module,
        "--fan-out",
        self.branch.fan_out_var,
        "--key",
        self.branch.key,
        *(("--cache-dir", self.cache_dir) if self.cache_dir is not None else ()),
        *(("--store-url", self.store_url) if self.store_url is not None else ()),
        "--params-json",
        self.params_json,
        "--expect-hash",
        self.branch.branch_hash,
    )

name

name() -> str

A job name unique per branch and attempt, for scheduler consoles.

Source code in src/adag/dispatch.py

def name(self) -> str:
    """A job name unique per branch and attempt, for scheduler consoles."""
    return f"adag-{self.branch.branch_hash}-{self.attempt}"

JobState

Bases: Enum

Where a submitted job is. SUCCEEDED and FAILED are final.

Source code in src/adag/dispatch.py

class JobState(Enum):
    """Where a submitted job is. ``SUCCEEDED`` and ``FAILED`` are final."""

    PENDING = "pending"
    RUNNING = "running"
    SUCCEEDED = "succeeded"
    FAILED = "failed"

JobStatus dataclass

A job's state, with detail (an error or exit code) for FAILED.

exit_code is set when the backend knows it. EXIT_BAD_BRANCH is never retried.

Source code in src/adag/dispatch.py

@dataclass(frozen=True)
class JobStatus:
    """A job's state, with ``detail`` (an error or exit code) for ``FAILED``.

    ``exit_code`` is set when the backend knows it. ``EXIT_BAD_BRANCH`` is never retried.
    """

    state: JobState
    detail: str = ""
    exit_code: int | None = None

JobBackend

Bases: Protocol

Runs branch jobs on some scheduler: a local process, AWS Batch, Kubernetes, SLURM.

RemoteDispatcher owns the cache check, retries, timeouts and cancel. A backend only starts jobs and reports on them. It must turn off the scheduler's own retries, or the attempts multiply.

Source code in src/adag/dispatch.py

class JobBackend(Protocol):
    """Runs branch jobs on some scheduler: a local process, AWS Batch, Kubernetes, SLURM.

    ``RemoteDispatcher`` owns the cache check, retries, timeouts and cancel. A backend only
    starts jobs and reports on them. It must turn off the scheduler's own retries, or the
    attempts multiply.
    """

    def submit(self, job: BranchJob) -> str:
        """Start ``job`` and return its job id."""
        ...

    def status(self, job_ids: Sequence[str]) -> Mapping[str, JobStatus]:
        """Report every job in ``job_ids``, in one call to the scheduler where it can."""
        ...

    def cancel(self, job_ids: Sequence[str]) -> None:
        """Stop the jobs. Jobs already finished are ignored."""
        ...

submit

submit(job: BranchJob) -> str

Start job and return its job id.

Source code in src/adag/dispatch.py

def submit(self, job: BranchJob) -> str:
    """Start ``job`` and return its job id."""
    ...

status

status(job_ids: Sequence[str]) -> Mapping[str, JobStatus]

Report every job in job_ids, in one call to the scheduler where it can.

Source code in src/adag/dispatch.py

def status(self, job_ids: Sequence[str]) -> Mapping[str, JobStatus]:
    """Report every job in ``job_ids``, in one call to the scheduler where it can."""
    ...

cancel

cancel(job_ids: Sequence[str]) -> None

Stop the jobs. Jobs already finished are ignored.

Source code in src/adag/dispatch.py

def cancel(self, job_ids: Sequence[str]) -> None:
    """Stop the jobs. Jobs already finished are ignored."""
    ...

RemoteDispatcher

Run each fan-out branch as a job on backend and load its result from the cache.

The dispatcher skips branches already in the cache, submits the rest (at most max_concurrent at a time), polls backend.status once per policy.poll_interval_seconds for all live jobs, retries failed attempts up to policy.max_retries, and cancels every live job when it stops early (an error, a timeout, or Ctrl-C). A job that exits with EXIT_BAD_BRANCH is not retried.

When the pipeline has a store, pass store_url (see store_from_url): a store object cannot cross to a worker, so the worker opens the same store from the URL.

clock and sleep are injectable so tests run without waiting.

Source code in src/adag/dispatch.py

class RemoteDispatcher:
    """Run each fan-out branch as a job on ``backend`` and load its result from the cache.

    The dispatcher skips branches already in the cache, submits the rest (at most
    ``max_concurrent`` at a time), polls ``backend.status`` once per
    ``policy.poll_interval_seconds`` for all live jobs, retries failed attempts up to
    ``policy.max_retries``, and cancels every live job when it stops early (an error, a
    timeout, or Ctrl-C). A job that exits with ``EXIT_BAD_BRANCH`` is not retried.

    When the pipeline has a ``store``, pass ``store_url`` (see ``store_from_url``): a store
    object cannot cross to a worker, so the worker opens the same store from the URL.

    ``clock`` and ``sleep`` are injectable so tests run without waiting.
    """

    def __init__(
        self,
        backend: JobBackend,
        policy: BranchPolicy | None = None,
        resources: Resources | None = None,
        max_concurrent: int | None = None,
        store_url: str | None = None,
        *,
        clock: Callable[[], float] = time.monotonic,
        sleep: Callable[[float], None] = time.sleep,
    ) -> None:
        self.backend = backend
        self.policy = policy or BranchPolicy()
        self.resources = resources or Resources()
        self.max_concurrent = max_concurrent
        self.store_url = store_url
        self._clock = clock
        self._sleep = sleep

    def dispatch(self, branch: BranchRef, sub: StepResult[Any], context: ResolveContext) -> Any:
        """Resolve one branch. Kept from the pre-0.2.11 one-branch protocol; calls ``dispatch_all``."""
        return self.dispatch_all([(branch, sub)], context)[branch.key]

    def dispatch_all(
        self, branches: Sequence[tuple[BranchRef, StepResult[Any]]], context: ResolveContext
    ) -> dict[str, Any]:
        """Resolve every branch on the backend; return ``{key: result}``.

        Raises:
            ValueError: the run has neither ``cache_dir`` nor ``store``, or it has a ``store``
                and ``store_url`` was not given.
            TypeError: a param is not JSON-serializable.
            FanOutBranchError: one or more branches failed. Results of the others are in
                the cache, so a rerun submits only the failed keys.
        """
        from adag.fan_out import FanOutBranchError

        store = context.result_store()
        if store is None:
            raise ValueError(
                f"{type(self).__name__} requires a cache_dir or store (the shared rendezvous)."
            )
        if context.store is not None and self.store_url is None:
            raise ValueError(
                f"{type(self).__name__} needs store_url= when the pipeline has a store, so "
                "workers can open the same store."
            )
        cache_dir = str(context.cache_dir) if context.store is None else None
        payload = params_json(context.params)

        results: dict[str, Any] = {}
        failed: dict[str, str] = {}
        now = self._clock()
        queue: deque[_Queued] = deque()
        for ref, sub in branches:
            if sub.is_cached(store, params=context.params):
                send_event(
                    context.observer, "branch.cache_hit", {"key": ref.key, "hash": ref.branch_hash}
                )
                results[ref.key] = sub.load(store, params=context.params)
            else:
                queue.append(_Queued(ref, sub, 1, now))

        live: dict[str, _LiveJob] = {}

        def fail(job_id: str, job: _LiveJob, detail: str, exit_code: int | None) -> None:
            del live[job_id]
            retryable = exit_code != EXIT_BAD_BRANCH and job.attempt <= self.policy.max_retries
            if retryable:
                ready_at = self._clock() + self.policy.retry_delay_seconds
                queue.append(_Queued(job.ref, job.sub, job.attempt + 1, ready_at))
                return
            failed[job.ref.key] = detail
            context.observer.on_fan_out_branch_failed(job.ref.key, detail)

        try:
            while queue or live:
                self._submit_ready(queue, live, payload, cache_dir, context)
                if live:
                    statuses = self.backend.status(list(live))
                    now = self._clock()
                    for job_id, job in list(live.items()):
                        status = statuses.get(job_id, JobStatus(JobState.PENDING))
                        self._advance(job_id, job, status, now, store, live, results, fail, context)
                if queue or live:
                    self._sleep(self.policy.poll_interval_seconds)
        finally:
            if live:
                self.backend.cancel(list(live))

        if failed:
            raise FanOutBranchError(succeeded=sorted(results), failed=failed)
        return results

    def _submit_ready(
        self,
        queue: deque[_Queued],
        live: dict[str, _LiveJob],
        payload: str,
        cache_dir: str | None,
        context: ResolveContext,
    ) -> None:
        """Submit queued attempts whose retry delay has passed, up to ``max_concurrent``."""
        now = self._clock()
        for _ in range(len(queue)):
            if self.max_concurrent is not None and len(live) >= self.max_concurrent:
                return
            item = queue.popleft()
            if item.ready_at > now:
                queue.append(item)
                continue
            job = BranchJob(
                item.ref, payload, cache_dir, item.attempt, self.resources, self.store_url
            )
            send_event(
                context.observer,
                "branch.dispatch",
                {"key": item.ref.key, "attempt": item.attempt, "of": self.policy.max_retries + 1},
            )
            job_id = self.backend.submit(job)
            live[job_id] = _LiveJob(item.ref, item.sub, item.attempt, now)

    def _advance(
        self,
        job_id: str,
        job: _LiveJob,
        status: JobStatus,
        now: float,
        store: CacheStore,
        live: dict[str, _LiveJob],
        results: dict[str, Any],
        fail: Callable[[str, _LiveJob, str, int | None], None],
        context: ResolveContext,
    ) -> None:
        """Move one live job forward from its latest status."""
        policy = self.policy
        if status.state is JobState.SUCCEEDED:
            if job.sub.is_cached(store, params=context.params):
                del live[job_id]
                results[job.ref.key] = job.sub.load(store, params=context.params)
                return
            if job.succeeded_at is None:
                job.succeeded_at = now
            if now - job.succeeded_at >= policy.result_grace_seconds:
                fail(job_id, job, "job succeeded but no result is in the cache", None)
            return
        if status.state is JobState.FAILED:
            fail(job_id, job, status.detail or "job failed", status.exit_code)
            return
        if status.state is JobState.RUNNING:
            if job.running_since is None:
                job.running_since = now
            timeout = policy.timeout_seconds
            if timeout is not None and now - job.running_since > timeout:
                self.backend.cancel([job_id])
                logger.warning("branch.timeout", key=job.ref.key, attempt=job.attempt)
                fail(job_id, job, f"timed out after {timeout}s", None)
            return
        queue_timeout = policy.queue_timeout_seconds
        if queue_timeout is not None and now - job.submitted_at > queue_timeout:
            self.backend.cancel([job_id])
            fail(job_id, job, f"still queued after {queue_timeout}s", None)

dispatch

dispatch(
    branch: BranchRef,
    sub: StepResult[Any],
    context: ResolveContext,
) -> Any

Resolve one branch. Kept from the pre-0.2.11 one-branch protocol; calls dispatch_all.

Source code in src/adag/dispatch.py

def dispatch(self, branch: BranchRef, sub: StepResult[Any], context: ResolveContext) -> Any:
    """Resolve one branch. Kept from the pre-0.2.11 one-branch protocol; calls ``dispatch_all``."""
    return self.dispatch_all([(branch, sub)], context)[branch.key]

dispatch_all

dispatch_all(
    branches: Sequence[tuple[BranchRef, StepResult[Any]]],
    context: ResolveContext,
) -> dict[str, Any]

Resolve every branch on the backend; return {key: result}.

RAISES	DESCRIPTION
ValueError	the run has neither cache_dir nor store, or it has a store and store_url was not given.
TypeError	a param is not JSON-serializable.
FanOutBranchError	one or more branches failed. Results of the others are in the cache, so a rerun submits only the failed keys.

Source code in src/adag/dispatch.py

def dispatch_all(
    self, branches: Sequence[tuple[BranchRef, StepResult[Any]]], context: ResolveContext
) -> dict[str, Any]:
    """Resolve every branch on the backend; return ``{key: result}``.

    Raises:
        ValueError: the run has neither ``cache_dir`` nor ``store``, or it has a ``store``
            and ``store_url`` was not given.
        TypeError: a param is not JSON-serializable.
        FanOutBranchError: one or more branches failed. Results of the others are in
            the cache, so a rerun submits only the failed keys.
    """
    from adag.fan_out import FanOutBranchError

    store = context.result_store()
    if store is None:
        raise ValueError(
            f"{type(self).__name__} requires a cache_dir or store (the shared rendezvous)."
        )
    if context.store is not None and self.store_url is None:
        raise ValueError(
            f"{type(self).__name__} needs store_url= when the pipeline has a store, so "
            "workers can open the same store."
        )
    cache_dir = str(context.cache_dir) if context.store is None else None
    payload = params_json(context.params)

    results: dict[str, Any] = {}
    failed: dict[str, str] = {}
    now = self._clock()
    queue: deque[_Queued] = deque()
    for ref, sub in branches:
        if sub.is_cached(store, params=context.params):
            send_event(
                context.observer, "branch.cache_hit", {"key": ref.key, "hash": ref.branch_hash}
            )
            results[ref.key] = sub.load(store, params=context.params)
        else:
            queue.append(_Queued(ref, sub, 1, now))

    live: dict[str, _LiveJob] = {}

    def fail(job_id: str, job: _LiveJob, detail: str, exit_code: int | None) -> None:
        del live[job_id]
        retryable = exit_code != EXIT_BAD_BRANCH and job.attempt <= self.policy.max_retries
        if retryable:
            ready_at = self._clock() + self.policy.retry_delay_seconds
            queue.append(_Queued(job.ref, job.sub, job.attempt + 1, ready_at))
            return
        failed[job.ref.key] = detail
        context.observer.on_fan_out_branch_failed(job.ref.key, detail)

    try:
        while queue or live:
            self._submit_ready(queue, live, payload, cache_dir, context)
            if live:
                statuses = self.backend.status(list(live))
                now = self._clock()
                for job_id, job in list(live.items()):
                    status = statuses.get(job_id, JobStatus(JobState.PENDING))
                    self._advance(job_id, job, status, now, store, live, results, fail, context)
            if queue or live:
                self._sleep(self.policy.poll_interval_seconds)
    finally:
        if live:
            self.backend.cancel(list(live))

    if failed:
        raise FanOutBranchError(succeeded=sorted(results), failed=failed)
    return results

LocalJobBackend

Run each branch job as a local python -m adag resolve-branch process.

stderr goes to a temporary file, so a chatty worker cannot block on a full pipe. Its last 500 characters become the detail of a FAILED status.

Source code in src/adag/dispatch.py

class LocalJobBackend:
    """Run each branch job as a local ``python -m adag resolve-branch`` process.

    stderr goes to a temporary file, so a chatty worker cannot block on a full pipe. Its last
    500 characters become the ``detail`` of a FAILED status.
    """

    def __init__(self) -> None:
        self._procs: dict[str, tuple[subprocess.Popen[bytes], IO[bytes]]] = {}
        self._next = 0

    def submit(self, job: BranchJob) -> str:
        # Closed in status() once the process ends, or in cancel().
        stderr = tempfile.TemporaryFile()  # noqa: SIM115
        proc = subprocess.Popen(
            job.command(prefix=(sys.executable, "-m", "adag")),
            stdout=subprocess.DEVNULL,
            stderr=stderr,
        )
        self._next += 1
        job_id = f"local-{self._next}"
        self._procs[job_id] = (proc, stderr)
        return job_id

    def status(self, job_ids: Sequence[str]) -> dict[str, JobStatus]:
        statuses: dict[str, JobStatus] = {}
        for job_id in job_ids:
            proc, stderr = self._procs[job_id]
            code = proc.poll()
            if code is None:
                statuses[job_id] = JobStatus(JobState.RUNNING)
                continue
            if code == 0:
                statuses[job_id] = JobStatus(JobState.SUCCEEDED, exit_code=0)
            else:
                stderr.seek(0)
                tail = stderr.read().decode(errors="replace").strip()[-500:]
                statuses[job_id] = JobStatus(
                    JobState.FAILED, tail or f"exit code {code}", exit_code=code
                )
            stderr.close()
            del self._procs[job_id]
        return statuses

    def cancel(self, job_ids: Sequence[str]) -> None:
        for job_id in job_ids:
            entry = self._procs.pop(job_id, None)
            if entry is None:
                continue
            proc, stderr = entry
            proc.terminate()
            try:
                proc.wait(timeout=5)
            except subprocess.TimeoutExpired:
                proc.kill()
            stderr.close()

SubprocessDispatcher

Bases: RemoteDispatcher

Resolve each branch in a local OS process via adag resolve-branch.

RemoteDispatcher with LocalJobBackend. Runs at most max_concurrent processes at once (default: the number of CPUs). Workers share results through cache_dir, or the store at store_url. The pipeline must have one of them.

Source code in src/adag/dispatch.py

class SubprocessDispatcher(RemoteDispatcher):
    """Resolve each branch in a local OS process via ``adag resolve-branch``.

    ``RemoteDispatcher`` with ``LocalJobBackend``. Runs at most ``max_concurrent`` processes at
    once (default: the number of CPUs). Workers share results through ``cache_dir``, or the
    store at ``store_url``. The pipeline must have one of them.
    """

    def __init__(
        self,
        policy: BranchPolicy | None = None,
        max_concurrent: int | None = None,
        store_url: str | None = None,
    ) -> None:
        super().__init__(
            LocalJobBackend(),
            policy or BranchPolicy(poll_interval_seconds=0.05),
            max_concurrent=max_concurrent or os.cpu_count() or 1,
            store_url=store_url,
        )

params_json

params_json(params: Mapping[str, Any]) -> str

Render run params as one JSON object, so the worker gets the same types.

RAISES	DESCRIPTION
TypeError	a param value is not JSON-serializable, so it cannot reach a worker.

Source code in src/adag/dispatch.py

def params_json(params: Mapping[str, Any]) -> str:
    """Render run params as one JSON object, so the worker gets the same types.

    Raises:
        TypeError: a param value is not JSON-serializable, so it cannot reach a worker.
    """
    try:
        return json.dumps(params, sort_keys=True)
    except TypeError as error:
        raise TypeError(f"dispatch params must be JSON-serializable: {error}") from error

Cache stores

adag.cache_store

CacheStore: the key-to-text backend behind every adag cache.

Step results, BatchStep item caches and dispatched fan-out branches are all stored as text under a key. Pass a store with Pipeline(..., store=). Without one, the pipeline uses FileSystemCacheStore on cache_dir.

A backend implements five methods over str values: get, set, exists, delete, iter_keys. ItemCache builds its schema stamp and in-flight markers from those five, so a backend does not need to know about them.

Stores with no extra install:

• FileSystemCacheStore: one text file per key, atomic writes. The default.
• SqliteCacheStore: one SQLite file, on stdlib sqlite3.
• MemoryCacheStore: a dict in this process, for tests.

Stores behind an extra (the driver is imported only when the store is built):

• SqlCacheStore: any SQLAlchemy database. Extra: adag[sql].
• RedisCacheStore: Redis or Valkey. Extra: adag[redis].

For dispatch, every worker and the main process must see the same store: Redis, a SQL database, or a filesystem all of them mount. A disk local to one machine does not work.

CacheStore

Bases: Protocol

A string key to string value store: the five methods a cache backend implements.

Implement them over any storage: Redis, a database, S3, memory.

Keys are opaque strings the cache layer builds (content hashes, namespaced like "namespace/abc123"). Values are UTF-8 text (the cache layer JSON-encodes its own payloads). get returns None for a missing key (never raises for absence).

Source code in src/adag/cache_store.py

@runtime_checkable
class CacheStore(Protocol):
    """A string key to string value store: the five methods a cache backend implements.

    Implement them over any storage: Redis, a database, S3, memory.

    Keys are opaque strings the cache layer builds (content hashes, namespaced like
    ``"namespace/abc123"``). Values are UTF-8 text (the cache layer JSON-encodes its
    own payloads). ``get`` returns ``None`` for a missing key (never raises for absence).
    """

    def get(self, key: str) -> str | None:
        """Return the stored text for ``key``, or ``None`` if absent."""
        ...

    def set(self, key: str, value: str) -> None:
        """Store ``value`` under ``key`` (overwrites). Should be atomic per key."""
        ...

    def exists(self, key: str) -> bool:
        """True if ``key`` has a stored value."""
        ...

    def delete(self, key: str) -> None:
        """Remove ``key``. A no-op if it is absent (never raises for absence)."""
        ...

    def iter_keys(self, prefix: str = "") -> Iterable[str]:
        """Yield every stored key starting with ``prefix`` (all keys if empty)."""
        ...

get

get(key: str) -> str | None

Return the stored text for key, or None if absent.

Source code in src/adag/cache_store.py

def get(self, key: str) -> str | None:
    """Return the stored text for ``key``, or ``None`` if absent."""
    ...

set

set(key: str, value: str) -> None

Store value under key (overwrites). Should be atomic per key.

Source code in src/adag/cache_store.py

def set(self, key: str, value: str) -> None:
    """Store ``value`` under ``key`` (overwrites). Should be atomic per key."""
    ...

exists

exists(key: str) -> bool

True if key has a stored value.

Source code in src/adag/cache_store.py

def exists(self, key: str) -> bool:
    """True if ``key`` has a stored value."""
    ...

delete

delete(key: str) -> None

Remove key. A no-op if it is absent (never raises for absence).

Source code in src/adag/cache_store.py

def delete(self, key: str) -> None:
    """Remove ``key``. A no-op if it is absent (never raises for absence)."""
    ...

iter_keys

iter_keys(prefix: str = '') -> Iterable[str]

Yield every stored key starting with prefix (all keys if empty).

Source code in src/adag/cache_store.py

def iter_keys(self, prefix: str = "") -> Iterable[str]:
    """Yield every stored key starting with ``prefix`` (all keys if empty)."""
    ...

FileSystemCacheStore

Stores one text file per key under root. The default.

Keys may contain / to nest into sub-directories. Writes are atomic (temp file, then rename).

Source code in src/adag/cache_store.py

class FileSystemCacheStore:
    """Stores one text file per key under ``root``. The default.

    Keys may contain ``/`` to nest into sub-directories. Writes are atomic (temp file, then
    rename).
    """

    def __init__(self, root: str | Path) -> None:
        self.root = Path(root)

    def _path(self, key: str) -> Path:
        if ".." in Path(key).parts:
            raise ValueError(f"unsafe cache key (path traversal): {key!r}")
        return self.root.joinpath(key)

    def get(self, key: str) -> str | None:
        path = self._path(key)
        try:
            return path.read_text()
        except (FileNotFoundError, IsADirectoryError):
            # A directory is a step result written by save_result, not a text value.
            return None

    def set(self, key: str, value: str) -> None:
        path = self._path(key)
        path.parent.mkdir(parents=True, exist_ok=True)
        atomic_write_text(path, value)

    def exists(self, key: str) -> bool:
        return self._path(key).exists()

    def delete(self, key: str) -> None:
        self._path(key).unlink(missing_ok=True)

    def iter_keys(self, prefix: str = "") -> Iterable[str]:
        if not self.root.exists():
            return
        for p in self.root.rglob("*"):
            if p.is_file() and not self._inside_path_result(p):
                key = p.relative_to(self.root).as_posix()
                if key.startswith(prefix):
                    yield key

    def _inside_path_result(self, path: Path) -> bool:
        """Return whether ``path`` is inside a directory result (a sibling ``<dir>.done`` exists)."""
        for parent in path.relative_to(self.root).parents:
            if parent != Path(".") and self.root.joinpath(f"{parent}.done").exists():
                return True
        return False

RedisCacheStore

Stores keys in Redis or Valkey, under key_prefix. Needs adag[redis].

The prefix lets one Redis host many apps. redis is imported lazily: only building this class (or passing a client) needs the package installed.

Pass an existing client (RedisCacheStore(client=...)) or a URL (RedisCacheStore(url="redis://…")). ttl_seconds optionally expires entries (None = persist; the DAG cache is content-addressed, so stale entries can't be wrong; TTL only controls eviction and cost).

Source code in src/adag/cache_store.py

class RedisCacheStore:
    """Stores keys in Redis or Valkey, under ``key_prefix``. Needs ``adag[redis]``.

    The prefix lets one Redis host many apps. ``redis`` is imported lazily: only building
    this class (or passing a client) needs the package installed.

    Pass an existing client (``RedisCacheStore(client=...)``) or a URL
    (``RedisCacheStore(url="redis://…")``). ``ttl_seconds`` optionally expires entries
    (None = persist; the DAG cache is content-addressed, so stale entries can't be
    wrong; TTL only controls eviction and cost).
    """

    def __init__(
        self,
        *,
        client: _RedisLike | None = None,
        url: str | None = None,
        key_prefix: str = "dag:",
        ttl_seconds: int | None = None,
    ) -> None:
        resolved: _RedisLike
        if client is None:
            try:
                # Optional backend dep, imported lazily; pyright can't resolve it when
                # it isn't installed, which is a supported case (filesystem is default).
                import redis  # pyright: ignore[reportMissingImports]
            except ImportError as exc:  # pragma: no cover - exercised only without redis installed
                raise ImportError(
                    "RedisCacheStore needs the 'redis' package: pip install redis"
                ) from exc
            # reportAssignmentType: when redis IS installed, redis-py's own typing makes the concrete
            # Redis structurally incompatible with _RedisLike (its get/set bind through an invariant
            # _is_async_client), so the protocol decoupling needs this ignore too.
            resolved = redis.Redis.from_url(url or "redis://localhost:6379/0")  # pyright: ignore[reportUnknownMemberType, reportUnknownArgumentType, reportAttributeAccessIssue, reportUnknownVariableType, reportAssignmentType]
        else:
            resolved = client
        self._client: _RedisLike = resolved
        self.key_prefix = key_prefix
        self.ttl_seconds = ttl_seconds

    def _k(self, key: str) -> str:
        return f"{self.key_prefix}{key}"

    @staticmethod
    def _text(raw: object) -> str:
        return raw.decode() if isinstance(raw, (bytes, bytearray)) else str(raw)

    def get(self, key: str) -> str | None:
        raw = self._client.get(self._k(key))
        return None if raw is None else self._text(raw)

    def set(self, key: str, value: str) -> None:
        self._client.set(self._k(key), value, ex=self.ttl_seconds)

    def exists(self, key: str) -> bool:
        return bool(self._client.exists(self._k(key)))

    def delete(self, key: str) -> None:
        self._client.delete(self._k(key))

    def iter_keys(self, prefix: str = "") -> Iterable[str]:
        match = f"{self.key_prefix}{prefix}*"
        for raw in self._client.scan_iter(match=match):
            yield self._text(raw)[len(self.key_prefix) :]

MemoryCacheStore

Stores keys in a dict in this process. Cannot be shared with a worker.

Use it for tests and single-process runs that should not touch disk.

Source code in src/adag/cache_store.py

class MemoryCacheStore:
    """Stores keys in a dict in this process. Cannot be shared with a worker.

    Use it for tests and single-process runs that should not touch disk.
    """

    def __init__(self) -> None:
        self._d: dict[str, str] = {}

    def get(self, key: str) -> str | None:
        return self._d.get(key)

    def set(self, key: str, value: str) -> None:
        self._d[key] = value

    def exists(self, key: str) -> bool:
        return key in self._d

    def delete(self, key: str) -> None:
        self._d.pop(key, None)

    def iter_keys(self, prefix: str = "") -> Iterable[str]:
        return [k for k in self._d if k.startswith(prefix)]

SqliteCacheStore

Stores keys in one SQLite file, on stdlib sqlite3.

One row per key in a cache(key, value) table; the DB is a single .db file, so it's a durable single-host cache without standing up Redis. Each op opens its own short-lived connection, so it's safe across the dispatch thread pool (SQLite's file locking serialises writers).

Right for a single-host service or a CLI run that wants the cache to survive a restart; NOT for heavy multi-instance write contention across hosts (use Redis/SQL there).

Source code in src/adag/cache_store.py

class SqliteCacheStore:
    """Stores keys in one SQLite file, on stdlib ``sqlite3``.

    One row per
    key in a ``cache(key, value)`` table; the DB is a single ``.db`` file, so it's a durable
    single-host cache without standing up Redis. Each op opens its own short-lived connection,
    so it's safe across the dispatch thread pool (SQLite's file locking serialises writers).

    Right for a single-host service or a CLI run that wants the cache to survive a restart;
    NOT for heavy multi-instance write contention across hosts (use Redis/SQL there).
    """

    def __init__(self, path: str | Path) -> None:
        self.path = Path(path)
        self.path.parent.mkdir(parents=True, exist_ok=True)
        with closing(sqlite3.connect(self.path)) as connection, connection:
            connection.execute(
                "CREATE TABLE IF NOT EXISTS cache (key TEXT PRIMARY KEY, value TEXT NOT NULL)"
            )

    def get(self, key: str) -> str | None:
        with closing(sqlite3.connect(self.path)) as connection:
            row: tuple[str] | None = connection.execute(
                "SELECT value FROM cache WHERE key = ?", (key,)
            ).fetchone()
        return row[0] if row is not None else None

    def set(self, key: str, value: str) -> None:
        with closing(sqlite3.connect(self.path)) as connection, connection:
            connection.execute(
                "INSERT INTO cache (key, value) VALUES (?, ?) ON CONFLICT(key) DO UPDATE SET value = excluded.value",
                (key, value),
            )

    def exists(self, key: str) -> bool:
        with closing(sqlite3.connect(self.path)) as connection:
            row: tuple[int] | None = connection.execute(
                "SELECT 1 FROM cache WHERE key = ?", (key,)
            ).fetchone()
        return row is not None

    def delete(self, key: str) -> None:
        with closing(sqlite3.connect(self.path)) as connection, connection:
            connection.execute("DELETE FROM cache WHERE key = ?", (key,))

    def iter_keys(self, prefix: str = "") -> Iterable[str]:
        with closing(sqlite3.connect(self.path)) as connection:
            rows: list[tuple[str]] = connection.execute(
                "SELECT key FROM cache ORDER BY key"
            ).fetchall()
        return [row[0] for row in rows if row[0].startswith(prefix)]

SqlCacheStore

Stores keys in one table of any SQLAlchemy database. Needs adag[sql].

Works with any URL SQLAlchemy supports (postgresql://…, mysql://…, sqlite:///path.db), so the DAG cache can live in the same database the rest of a deployed service already uses. sqlalchemy is a lazily-imported optional dependency.

One key/value table (name configurable). set overwrites via delete-then-insert in one transaction, which works on every dialect without a dialect-specific UPSERT. Pass a URL (SqlCacheStore(url="postgresql://…")) or an existing engine (SqlCacheStore(engine=…)) to share a connection pool with the host app.

Source code in src/adag/cache_store.py

class SqlCacheStore:
    """Stores keys in one table of any SQLAlchemy database. Needs ``adag[sql]``.

    Works with any URL SQLAlchemy supports (``postgresql://…``,
    ``mysql://…``, ``sqlite:///path.db``), so the DAG cache can live in the same database the
    rest of a deployed service already uses. ``sqlalchemy`` is a lazily-imported optional
    dependency.

    One ``key/value`` table (name configurable). ``set`` overwrites via delete-then-insert in
    one transaction, which works on every dialect without a dialect-specific UPSERT. Pass a URL
    (``SqlCacheStore(url="postgresql://…")``) or an existing engine (``SqlCacheStore(engine=…)``)
    to share a connection pool with the host app.
    """

    def __init__(
        self, *, url: str | None = None, engine: Engine | None = None, table_name: str = "dag_cache"
    ) -> None:
        try:
            from sqlalchemy import Column, MetaData, String, Table, create_engine
        except ImportError as exc:  # pragma: no cover - exercised only without sqlalchemy installed
            raise ImportError(
                "SqlCacheStore needs the 'sqlalchemy' package: pip install sqlalchemy"
            ) from exc
        if engine is None:
            if url is None:
                raise ValueError("SqlCacheStore requires either 'url' or 'engine'")
            engine = create_engine(url)
        self._engine: Engine = engine
        self._table = Table(
            table_name,
            MetaData(),
            Column("key", String, primary_key=True),
            Column("value", String, nullable=False),
        )
        self._table.metadata.create_all(engine)

    def get(self, key: str) -> str | None:
        from sqlalchemy import select

        with self._engine.connect() as connection:
            value: str | None = connection.execute(
                select(self._table.c.value).where(self._table.c.key == key)
            ).scalar_one_or_none()
        return value

    def set(self, key: str, value: str) -> None:
        from sqlalchemy import delete, insert

        with self._engine.begin() as connection:
            connection.execute(delete(self._table).where(self._table.c.key == key))
            connection.execute(insert(self._table).values(key=key, value=value))

    def exists(self, key: str) -> bool:
        from sqlalchemy import select

        with self._engine.connect() as connection:
            row = connection.execute(
                select(self._table.c.key).where(self._table.c.key == key)
            ).first()
        return row is not None

    def delete(self, key: str) -> None:
        from sqlalchemy import delete

        with self._engine.begin() as connection:
            connection.execute(delete(self._table).where(self._table.c.key == key))

    def iter_keys(self, prefix: str = "") -> Iterable[str]:
        from sqlalchemy import select

        with self._engine.connect() as connection:
            keys: list[str] = list(
                connection.execute(select(self._table.c.key).order_by(self._table.c.key))
                .scalars()
                .all()
            )
        return [key for key in keys if key.startswith(prefix)]

store_from_url

store_from_url(url: str) -> CacheStore

Open a CacheStore from a URL, so a remote worker can reach the main process's store.

• file:///path or a plain path: FileSystemCacheStore.
• sqlite:///relative.db or sqlite:////absolute.db: SqliteCacheStore (stdlib).
• redis://... or rediss://...: RedisCacheStore (needs adag[redis]).
• any other scheme://...: SqlCacheStore through SQLAlchemy (needs adag[sql]).

Credentials belong in the URL as the worker's environment provides it, never in params.

RAISES	DESCRIPTION
ValueError	memory: was given; a memory store cannot be shared with a worker.

Source code in src/adag/cache_store.py

def store_from_url(url: str) -> CacheStore:
    """Open a ``CacheStore`` from a URL, so a remote worker can reach the main process's store.

    - ``file:///path`` or a plain path: ``FileSystemCacheStore``.
    - ``sqlite:///relative.db`` or ``sqlite:////absolute.db``: ``SqliteCacheStore`` (stdlib).
    - ``redis://...`` or ``rediss://...``: ``RedisCacheStore`` (needs ``adag[redis]``).
    - any other ``scheme://...``: ``SqlCacheStore`` through SQLAlchemy (needs ``adag[sql]``).

    Credentials belong in the URL as the worker's environment provides it, never in params.

    Raises:
        ValueError: ``memory:`` was given; a memory store cannot be shared with a worker.
    """
    if url.startswith("memory:"):
        raise ValueError("a MemoryCacheStore cannot be shared with another process")
    if "://" not in url:
        return FileSystemCacheStore(url)
    scheme, _, rest = url.partition("://")
    if scheme == "file":
        return FileSystemCacheStore(rest)
    if scheme == "sqlite":
        # SQLAlchemy's form: sqlite:///relative.db and sqlite:////absolute.db
        return SqliteCacheStore(rest.removeprefix("/"))
    if scheme in {"redis", "rediss"}:
        return RedisCacheStore(url=url)
    return SqlCacheStore(url=url)

Hashing

adag.hashing

Content-addressable hashing utilities for the DAG engine.

KEY_FORMAT module-attribute

KEY_FORMAT = 2

Version of the cache key bytes. It changes only when a release changes every key on purpose.

1 is the first release; 2 is the framed encoding from 0.2.0. tests/test_key_golden.py pins the current format.

FileFingerprint dataclass

Identity of a file's content for caching: same fingerprint ⇒ same bytes (cheap stand-in for re-reading). Inode + device are part of the identity because mtime resolution on many filesystems is 1s, and an atomic-replace inside the same second produces a new inode but identical (path, mtime, size): without st_ino the cache would return the stale hash. Frozen ⇒ hashable ⇒ usable as a dict key.

Source code in src/adag/hashing.py

@dataclass(frozen=True)
class FileFingerprint:
    """Identity of a file's content for caching: same fingerprint ⇒ same bytes
    (cheap stand-in for re-reading). Inode + device are part of the identity
    because mtime resolution on many filesystems is 1s, and an atomic-replace
    inside the same second produces a new inode but identical (path, mtime,
    size): without st_ino the cache would return the stale hash. Frozen ⇒
    hashable ⇒ usable as a dict key.
    """

    path: str
    dev: int
    inode: int
    mtime: float
    size: int

UnhashableValueError

Bases: TypeError

A value has a type the cache key cannot encode.

Source code in src/adag/hashing.py

class UnhashableValueError(TypeError):
    """A value has a type the cache key cannot encode."""

file_hash

file_hash(path: Path) -> str

SHA-256 of file content, first 12 hex chars. Session-cached by file fingerprint.

Source code in src/adag/hashing.py

def file_hash(path: Path) -> str:
    """SHA-256 of file content, first 12 hex chars. Session-cached by file fingerprint."""
    resolved = path.resolve()
    stat = resolved.stat()
    cache_key = FileFingerprint(
        path=str(resolved),
        dev=stat.st_dev,
        inode=stat.st_ino,
        mtime=stat.st_mtime,
        size=stat.st_size,
    )
    cached = _file_hash_cache.get(cache_key)
    if cached is not None:
        return cached

    if stat.st_size > _LARGE_FILE_BYTES:
        from adag.deferred import emit  # deferred imports this module; import late

        emit("hashing.large_file", path=str(path), size_mb=round(stat.st_size / 1_000_000, 1))

    h = hashlib.sha256()
    with open(resolved, "rb") as f:
        while chunk := f.read(8192):
            h.update(chunk)
    result = h.hexdigest()[:12]
    _file_hash_cache[cache_key] = result
    return result

is_list

is_list(obj: Any) -> TypeGuard[list[Any]]

Type guard for list[Any]. Use instead of isinstance to avoid list[Unknown].

Source code in src/adag/hashing.py

def is_list(obj: Any) -> TypeGuard[list[Any]]:
    """Type guard for list[Any]. Use instead of isinstance to avoid list[Unknown]."""
    return isinstance(obj, list)

is_tuple

is_tuple(obj: Any) -> TypeGuard[tuple[Any, ...]]

Type guard for tuple[Any, ...]. Use instead of isinstance to avoid tuple[Unknown, ...].

Source code in src/adag/hashing.py

def is_tuple(obj: Any) -> TypeGuard[tuple[Any, ...]]:
    """Type guard for tuple[Any, ...]. Use instead of isinstance to avoid tuple[Unknown, ...]."""
    return isinstance(obj, tuple)

is_dict

is_dict(obj: Any) -> TypeGuard[dict[str, Any]]

Type guard for dict[str, Any]. Use instead of isinstance to avoid dict[Unknown, Unknown].

Source code in src/adag/hashing.py

def is_dict(obj: Any) -> TypeGuard[dict[str, Any]]:
    """Type guard for dict[str, Any]. Use instead of isinstance to avoid dict[Unknown, Unknown]."""
    return isinstance(obj, dict)

encode_value

encode_value(value: Any, where: str = 'value') -> bytes

Return the cache-key bytes for one value.

Every value is framed with a type tag and a byte length, so (1, 23) and (12, 3), or 1 and "1", never encode the same. A Path is hashed by its text, not its file content: use FileSource to hash content.

RAISES	DESCRIPTION
UnhashableValueError	value has no stable encoding. where names it in the message. Wrap execution-only values in Config(...); pass a graph node (a StepResult) as a call input instead.

Source code in src/adag/hashing.py

def encode_value(value: Any, where: str = "value") -> bytes:
    """Return the cache-key bytes for one value.

    Every value is framed with a type tag and a byte length, so ``(1, 23)`` and
    ``(12, 3)``, or ``1`` and ``"1"``, never encode the same. A ``Path`` is hashed by
    its text, not its file content: use ``FileSource`` to hash content.

    Raises:
        UnhashableValueError: ``value`` has no stable encoding. ``where`` names it in
            the message. Wrap execution-only values in ``Config(...)``; pass a graph node
            (a ``StepResult``) as a call input instead.
    """
    if value is None:
        return _frame(b"N", b"")
    if isinstance(value, bool):
        return _frame(b"B", b"1" if value else b"0")
    if isinstance(value, Enum):
        cls = type(value)
        name = f"{cls.__module__}.{cls.__qualname__}".encode()
        return _frame(b"E", _frame(b"S", name) + encode_value(value.value, where))
    if isinstance(value, int):
        return _frame(b"I", str(value).encode())
    if isinstance(value, float):
        return _frame(b"F", repr(value).encode())
    if isinstance(value, str):
        return _frame(b"S", value.encode())
    if isinstance(value, bytes):
        return _frame(b"Y", value)
    if isinstance(value, UUID):
        return _frame(b"Q", str(value).encode())
    if isinstance(value, PurePath):
        return _frame(b"P", value.as_posix().encode())
    if isinstance(value, type):
        return _frame(b"T", f"{value.__module__}.{value.__qualname__}".encode())
    if isinstance(value, BaseModel):
        cls = type(value)
        dumped = json.dumps(value.model_dump(mode="json"), sort_keys=True).encode()
        return _frame(b"M", f"{cls.__module__}.{cls.__qualname__}".encode() + b"\0" + dumped)
    return _encode_container(value, where)

hash_scalar

hash_scalar(value: Any) -> bytes

Return json.dumps(value, sort_keys=True) as bytes, for building your own item keys.

Not used by adag's own cache keys (those use encode_value). The bytes are the same as in the first release, so keys built with it stay valid.

Source code in src/adag/hashing.py

def hash_scalar(value: Any) -> bytes:
    """Return ``json.dumps(value, sort_keys=True)`` as bytes, for building your own item keys.

    Not used by adag's own cache keys (those use ``encode_value``). The bytes are the same as in
    the first release, so keys built with it stay valid.
    """
    return json.dumps(value, sort_keys=True).encode()

hash_list

hash_list(value: list[Any]) -> bytes

Return json.dumps(value, sort_keys=True) as bytes. Same bytes as hash_scalar.

Source code in src/adag/hashing.py

def hash_list(value: list[Any]) -> bytes:
    """Return ``json.dumps(value, sort_keys=True)`` as bytes. Same bytes as ``hash_scalar``."""
    return json.dumps(value, sort_keys=True).encode()

class_source_hash

class_source_hash(cls: type) -> bytes | None

Return a short hash of the class's own source code, or None if it has none.

Only the source of the class body is hashed: the first class in the MRO whose source inspect can read. Edits to a base class or to a helper function the step calls do NOT change the hash. Bump a field or kind version when such an edit changes results. Any edit inside the class body, comments included, does.

Memoized per class: inspect.getsource() re-reads + re-tokenizes the source file on every call, and this is invoked once per step instance across the whole graph, so caching turns a per-call file tokenization into a one-off.

Source code in src/adag/hashing.py

def class_source_hash(cls: type) -> bytes | None:
    """Return a short hash of the class's own source code, or None if it has none.

    Only the source of the class body is hashed: the first class in the MRO whose
    source ``inspect`` can read. Edits to a base class or to a helper function the
    step calls do NOT change the hash. Bump a field or ``kind`` version when such an
    edit changes results. Any edit inside the class body, comments included, does.

    Memoized per class: inspect.getsource() re-reads + re-tokenizes the source
    file on every call, and this is invoked once per step instance across the
    whole graph, so caching turns a per-call file tokenization into a one-off.
    """
    if cls in _class_source_hash_cache:
        return _class_source_hash_cache[cls]
    result: bytes | None = None
    for klass in cls.__mro__:
        if klass is object:
            break
        try:
            source = inspect.getsource(klass)
            result = hashlib.sha256(source.encode()).hexdigest()[:12].encode()
            break
        except (OSError, TypeError):
            continue
    _class_source_hash_cache[cls] = result
    return result