SPipe

steerability.spipe

.spipe: a portable serialization format for SteeringPipeline.

One format holds both the recipe (model reference plus controls-as-constructed) and, when frozen, the resolution (fingerprints, resolved bindings, per-fit digests, and a content-addressed artifact store). See docs/concepts/spipe.md for the model and the sharing guidance.

DataRef dataclass

A reference to a dataset, resolved at pipeline construction.

Attributes:

Name	Type	Description
kind	str	"hf" (a Hugging Face Hub dataset), "path" (a digest-pinned local file), or "opaque" (an in-memory dataset recorded by fingerprint only, not reloadable).
repo_id	str | None	Hub dataset id ("hf" only).
revision	str | None	Hub revision ("hf" only).
split	str | None	Dataset split ("hf" only).
subset	str | None	Dataset configuration name ("hf" only).
path	str | None	Local file path ("path" only).
sha256	str | None	Hex digest the file must match ("path" only).
type	str | None	Type qualname of the recorded object ("opaque" only).
fingerprint	str | None	Content fingerprint of the recorded object ("opaque" only), or None.

Source code in steerability/spipe/codec.py

@dataclass(frozen=True)
class DataRef:
    """A reference to a dataset, resolved at pipeline construction.

    Attributes:
        kind: `"hf"` (a Hugging Face Hub dataset), `"path"` (a digest-pinned local file), or
            `"opaque"` (an in-memory dataset recorded by fingerprint only, not reloadable).
        repo_id: Hub dataset id (`"hf"` only).
        revision: Hub revision (`"hf"` only).
        split: Dataset split (`"hf"` only).
        subset: Dataset configuration name (`"hf"` only).
        path: Local file path (`"path"` only).
        sha256: Hex digest the file must match (`"path"` only).
        type: Type qualname of the recorded object (`"opaque"` only).
        fingerprint: Content fingerprint of the recorded object (`"opaque"` only), or None.
    """

    kind: str
    repo_id: str | None = None
    revision: str | None = None
    split: str | None = None
    subset: str | None = None
    path: str | None = None
    sha256: str | None = None
    type: str | None = None
    fingerprint: str | None = None

    def load(self) -> Any:
        """Materialize the reference.

        Returns:
            The loaded dataset for `"hf"`, the verified path string for `"path"`.

        Raises:
            SpipeFormatError: If the kind is `"opaque"` (not reloadable) or unknown.
            SpipeIntegrityError: If a `"path"` file does not match its recorded digest.
        """
        if self.kind == "hf":
            from datasets import load_dataset

            kwargs: dict[str, Any] = {}
            if self.revision is not None:
                kwargs["revision"] = self.revision
            if self.split is not None:
                kwargs["split"] = self.split
            if self.subset is not None:
                return load_dataset(self.repo_id, self.subset, **kwargs)
            return load_dataset(self.repo_id, **kwargs)
        if self.kind == "path":
            import hashlib

            from steerability.spipe.errors import SpipeIntegrityError

            digest = hashlib.sha256(Path(self.path).read_bytes()).hexdigest()
            if self.sha256 is not None and digest != self.sha256:
                raise SpipeIntegrityError(
                    f"Data file {self.path} hashes to {digest}, expected {self.sha256}."
                )
            return self.path
        raise SpipeFormatError(
            f"DataRef kind {self.kind!r} cannot be materialized; an opaque dataset reference "
            "records identity only. Re-supply the dataset (or a 'hf'/'path' DataRef) to re-steer."
        )

fingerprint = None class-attribute instance-attribute

kind instance-attribute

path = None class-attribute instance-attribute

repo_id = None class-attribute instance-attribute

revision = None class-attribute instance-attribute

sha256 = None class-attribute instance-attribute

split = None class-attribute instance-attribute

subset = None class-attribute instance-attribute

type = None class-attribute instance-attribute

load()

Materialize the reference.

Returns:

Type	Description
Any	The loaded dataset for "hf", the verified path string for "path".

Raises:

Type	Description
SpipeFormatError	If the kind is "opaque" (not reloadable) or unknown.
SpipeIntegrityError	If a "path" file does not match its recorded digest.

Source code in steerability/spipe/codec.py

def load(self) -> Any:
    """Materialize the reference.

    Returns:
        The loaded dataset for `"hf"`, the verified path string for `"path"`.

    Raises:
        SpipeFormatError: If the kind is `"opaque"` (not reloadable) or unknown.
        SpipeIntegrityError: If a `"path"` file does not match its recorded digest.
    """
    if self.kind == "hf":
        from datasets import load_dataset

        kwargs: dict[str, Any] = {}
        if self.revision is not None:
            kwargs["revision"] = self.revision
        if self.split is not None:
            kwargs["split"] = self.split
        if self.subset is not None:
            return load_dataset(self.repo_id, self.subset, **kwargs)
        return load_dataset(self.repo_id, **kwargs)
    if self.kind == "path":
        import hashlib

        from steerability.spipe.errors import SpipeIntegrityError

        digest = hashlib.sha256(Path(self.path).read_bytes()).hexdigest()
        if self.sha256 is not None and digest != self.sha256:
            raise SpipeIntegrityError(
                f"Data file {self.path} hashes to {digest}, expected {self.sha256}."
            )
        return self.path
    raise SpipeFormatError(
        f"DataRef kind {self.kind!r} cannot be materialized; an opaque dataset reference "
        "records identity only. Re-supply the dataset (or a 'hf'/'path' DataRef) to re-steer."
    )

NotFreezableError

Bases: RuntimeError

A control produces steer-time state but declares no frozen form.

Raised by BaseControl.frozen_form when a control's steer step produces fits or exported state and the control does not override frozen_form to name a constructor-valid frozen form. Re-exported from steerability.spipe.errors alongside the other spipe exceptions.

Source code in steerability/algorithms/core/base_control.py

class NotFreezableError(RuntimeError):
    """A control produces steer-time state but declares no frozen form.

    Raised by `BaseControl.frozen_form` when a control's steer step produces fits or exported
    state and the control does not override `frozen_form` to name a constructor-valid frozen
    form. Re-exported from `steerability.spipe.errors` alongside the other spipe exceptions.
    """

SPipe

A serialized steering pipeline: recipe, optional lock, and artifact store.

Instances come from SteeringPipeline.to_spipe() (backed by a temporary store until saved) or SPipe.load() (backed by the loaded bundle). The manifest is data; controls are instantiated only by pipeline().

Source code in steerability/spipe/spipe.py

class SPipe:
    """A serialized steering pipeline: recipe, optional lock, and artifact store.

    Instances come from `SteeringPipeline.to_spipe()` (backed by a temporary store until
    saved) or `SPipe.load()` (backed by the loaded bundle). The manifest is data; controls
    are instantiated only by `pipeline()`.
    """

    def __init__(
        self,
        manifest: dict,
        *,
        store: ArtifactStore | None,
        base_dir: Path | None,
        allow_code: bool,
        allow_stale: bool = False,
        _temp: Any = None,
    ):
        validate_manifest(manifest)
        self._manifest = manifest
        self._manifest_records = _manifest_records(manifest)
        self._store = store
        self._base_dir = base_dir
        self._allow_code = allow_code
        self._allow_stale = allow_stale
        self._temp = _temp
        self._staleness_checked = False

    # inspection

    @property
    def manifest(self) -> dict:
        """A deep copy of the manifest."""
        return copy.deepcopy(self._manifest)

    @property
    def recipe_id(self) -> str:
        """The recipe identity digest (model reference plus controls), 12 hex characters."""
        lock = self._manifest.get("lock")
        if lock is not None:
            return lock["recipe_id"]
        from steerability.algorithms.core.identity import config_digest

        return config_digest({
            "model": self._manifest["model"]["ref"],
            "controls": self._descriptor()["controls"],
        })

    @property
    def config_id(self) -> str:
        """The configuration identity digest over the recipe entries, 12 hex characters."""
        lock = self._manifest.get("lock")
        if lock is not None:
            return lock["config_id"]
        from steerability.algorithms.core.identity import config_digest

        return config_digest(self._descriptor())

    def _descriptor(self) -> dict:
        """The configuration descriptor recomputed from freshly instantiated recipe controls."""
        from steerability.algorithms.core.identity import config_descriptor_from_controls

        controls = [self._instantiate(entry["method"], entry["args"], self._decode_ctx(lenient=True))
                    for entry in self._manifest["controls"]]
        for control, entry in zip(controls, self._manifest["controls"]):
            control.enabled = entry["enabled"]
        return config_descriptor_from_controls(controls)

    @property
    def code_dependent(self) -> bool:
        """Whether loading this spipe needs matching code and `allow_code=True`."""
        return bool(self._manifest["code_dependent"])

    @property
    def is_frozen(self) -> bool:
        """True iff every enabled entry with steer-time products carries a resolution.

        Entries with `resolved: null` have nothing to pin (their recipe is their frozen
        form), which makes a manifest with a lock section frozen.
        """
        return self._manifest.get("lock") is not None

    def describe(self) -> str:
        """A human-readable table of entries, frozen state, and artifact sizes."""
        lines = [
            f"spipe {self._manifest['format']}  model={self._manifest['model']['ref']}",
            f"  recipe_id={self.recipe_id}  config_id={self.config_id}  "
            f"frozen={self.is_frozen}  code_dependent={self.code_dependent}",
        ]
        for i, entry in enumerate(self._manifest["controls"]):
            items = _resolved_items(entry)
            frozen = "recipe"
            if items:
                methods = ", ".join(item["method"] for item in items)
                frozen = f"frozen -> {methods}"
            elif self.is_frozen and entry["enabled"]:
                frozen = "frozen (recipe is the frozen form)"
            enabled = "" if entry["enabled"] else "  [disabled]"
            lines.append(f"  [{i}] {entry['method']}{enabled}  ({frozen})")
            for item in items:
                for name, record in (item.get("artifacts") or {}).items():
                    size = ""
                    if self._store is not None and self._store.has(record["id"]):
                        size = f"  {self._store.size_of(record['id'])} bytes"
                    lines.append(f"        {name}: {record['type']} {record['id'][:19]}…{size}")
        return "\n".join(lines)

    # freeze state

    def thaw(self) -> "SPipe":
        """A recipe-only copy: every `resolved` section and the lock are dropped."""
        manifest = self.manifest
        for entry in manifest["controls"]:
            entry["resolved"] = None
        manifest["lock"] = None
        return SPipe(
            manifest, store=self._store, base_dir=self._base_dir,
            allow_code=self._allow_code, allow_stale=self._allow_stale, _temp=self._temp,
        )

    # verification

    def _referenced_artifact_ids(self) -> set[str]:
        found: set[str] = set()
        _collect_artifact_ids(self._manifest["controls"], found)
        return found

    def verify(self) -> SpipeReport:
        """The model-free report: format, integrity, staleness, versions, code dependence.

        Never loads a model or instantiates a backend. Referenced artifacts that are not
        present are reported in one warning (a thin bundle); the present ones are verified
        against their content ids.

        Returns:
            The report.
        """
        errors: list[str] = []
        report_warnings: list[str] = []

        try:
            validate_manifest(self._manifest)
        except SpipeFormatError as exc:
            errors.append(str(exc))

        referenced = sorted(self._referenced_artifact_ids())
        missing = [
            artifact_id for artifact_id in referenced
            if self._store is None or not self._store.has(artifact_id)
        ]
        if missing:
            report_warnings.append(
                f"{len(missing)} of {len(referenced)} referenced artifact(s) are not present (thin "
                f"bundle); pass artifact_store= at load to resolve them: {', '.join(missing)}."
            )
        for artifact_id in referenced:
            if artifact_id not in missing:
                try:
                    self._store.verify(artifact_id)
                except SpipeIntegrityError as exc:
                    errors.append(str(exc))

        try:
            self._check_staleness(raise_on_stale=True, force=True)
        except SpipeStaleError as exc:
            errors.append(str(exc))
        except Exception as exc:
            report_warnings.append(f"staleness check could not run: {exc}")

        from steerability.spipe.freeze import _toolkit_version

        saved = self._manifest.get("toolkit_version", "unknown")
        current = _toolkit_version()
        if saved.split(".")[0] != current.split(".")[0]:
            report_warnings.append(
                f"spipe was written by steerability {saved}; this is {current} (major mismatch)."
            )
        if self.code_dependent:
            report_warnings.append(
                "the manifest references code ($ref); loading a pipeline from it requires "
                "allow_code=True and the referenced modules on the import path."
            )

        return SpipeReport(ok=not errors, errors=errors, warnings=report_warnings)

    def _decode_ctx(self, *, lenient: bool, verify: str = "off", data_mode: str = "keep") -> DecodeContext:
        return DecodeContext(
            store=self._store,
            allow_code=self._allow_code,
            code_mode="sentinel" if lenient else "strict",
            verify=verify,
            data_mode=data_mode,
            manifest_records=self._manifest_records,
        )

    @staticmethod
    def _instantiate(method_key: str, encoded_args: Mapping, ctx: DecodeContext):
        from steerability.algorithms.core.registry import RegistryError, resolve_method_key

        try:
            method = resolve_method_key(method_key)
        except RegistryError as exc:
            raise SpipeFormatError(str(exc)) from exc
        kwargs = {name: decode(value, ctx, f"args.{name}") for name, value in encoded_args.items()}
        return method.control_cls(**kwargs)

    def _check_staleness(self, *, raise_on_stale: bool, force: bool = False) -> None:
        """Recompute each frozen entry's fit digest from the current recipe args.

        Skipped when artifacts are unavailable (thin bundle without a store); `pipeline()`
        re-invokes it once artifacts resolve.

        Raises:
            SpipeStaleError: If a recorded fit digest does not match the recomputed one, or a
                recipe entry can no longer be reconstructed for the check.
        """
        if self._staleness_checked and not force:
            return
        for i, entry in enumerate(self._manifest["controls"]):
            recorded: dict[str, str] = {}
            for item in _resolved_items(entry):
                for name, record in (item.get("artifacts") or {}).items():
                    if record.get("fit_digest"):
                        recorded[name] = record["fit_digest"]
            if not recorded:
                continue
            try:
                control = self._instantiate(entry["method"], entry["args"], self._decode_ctx(lenient=True))
                fit_identity = control.fit_identity()
            except Exception as exc:
                raise SpipeStaleError(
                    f"controls[{i}] ({entry['method']}): the recipe args no longer reconstruct "
                    f"for the staleness check ({exc}); thaw() and re-steer(), or pass "
                    "allow_stale=True."
                ) from exc
            current = digest_of(fit_identity) if fit_identity is not None else None
            for name, digest in recorded.items():
                if digest != current:
                    message = (
                        f"controls[{i}] ({entry['method']}): frozen artifact {name!r} was "
                        f"produced from fit digest {digest} but the recipe now digests to "
                        f"{current}; the fit-relevant recipe fields were edited after "
                        "freezing. thaw() and re-steer(), or pass allow_stale=True."
                    )
                    if raise_on_stale:
                        raise SpipeStaleError(message)
                    warnings.warn(message, UserWarning)
        self._staleness_checked = True

    # load / save

    @classmethod
    def load(
        cls,
        path: str | Path,
        *,
        allow_code: bool = False,
        allow_stale: bool = False,
        artifact_store: str | Path | Callable[[str], Path] | None = None,
    ) -> "SPipe":
        """Load a `.spipe` file or directory.

        Args:
            path: A `.spipe` zip file or a spipe directory.
            allow_code: Permit `$ref` imports, non-`steerability.` `$dc` imports, and
                pickle-backed memory payloads during decoding.
            allow_stale: Skip the staleness check.
            artifact_store: External artifact source for thin bundles: a store directory, or
                a callable mapping an artifact id to the directory holding it.

        Returns:
            The loaded `SPipe`.

        Raises:
            SpipeFormatError: On format-version, schema, or archive violations.
            SpipeIntegrityError: If a present artifact fails content verification.
            SpipeStaleError: If a frozen entry is stale and `allow_stale` is False.
        """
        path = Path(path)
        temp = None
        if path.is_dir():
            base_dir = path
        else:
            temp = tempfile.TemporaryDirectory(prefix="spipe-load-")
            unpack_zip(path, temp.name)
            base_dir = Path(temp.name)

        manifest = read_manifest(base_dir)

        resolver: Callable[[str], Path] | None = None
        if callable(artifact_store):
            resolver = artifact_store
        elif artifact_store is not None:
            external = Path(artifact_store)
            resolver = lambda artifact_id: external / artifact_id.replace(":", "-", 1)  # noqa: E731
        store = ArtifactStore(base_dir / ARTIFACTS_DIR, resolver=resolver)

        spipe = cls(
            manifest, store=store, base_dir=base_dir,
            allow_code=allow_code, allow_stale=allow_stale, _temp=temp,
        )

        referenced = spipe._referenced_artifact_ids()
        available = [artifact_id for artifact_id in referenced if store.has(artifact_id)]
        for artifact_id in sorted(available):
            store.verify(artifact_id)
        if len(available) < len(referenced):
            logger.info(
                "Thin spipe: %d of %d referenced artifacts unavailable; integrity and "
                "staleness checks defer to pipeline construction.",
                len(referenced) - len(available), len(referenced),
            )
        elif not allow_stale:
            spipe._check_staleness(raise_on_stale=True)

        return spipe

    def save(self, path: str | Path, *, artifacts: str = "fat") -> Path:
        """Write the spipe to `path`.

        A path ending in `.spipe` produces a zip file; any other path produces (or replaces)
        a directory. `artifacts="fat"` (default) embeds every referenced artifact;
        `artifacts="thin"` writes the manifest only, leaving artifact ids resolvable at load
        via `artifact_store=`. A recipe-only spipe with no artifact references writes no
        `artifacts/` directory either way.

        Saving onto the bundle's own backing directory rewrites the manifest in place. A fat
        save there also embeds any referenced artifact the store resolves externally, and a
        thin save there raises `SpipeSaveError` when the directory embeds artifacts, since a
        thin export would have to delete them.

        Args:
            path: Destination file or directory.
            artifacts: `"fat"` or `"thin"`.

        Returns:
            The written path.

        Raises:
            SpipeSaveError: If a directory target exists and is neither empty nor a spipe
                directory, a referenced artifact is unavailable for a fat export, or a thin
                export targets the bundle's own directory while it embeds artifacts.
        """
        if artifacts not in ("fat", "thin"):
            raise SpipeSaveError(f"artifacts must be 'fat' or 'thin'; got {artifacts!r}.")
        path = Path(path)
        referenced = sorted(self._referenced_artifact_ids())
        if artifacts == "fat" and referenced:
            if self._store is None:
                raise SpipeSaveError("This spipe has no artifact store; only artifacts='thin' is possible.")
            for artifact_id in referenced:
                self._store.verify(artifact_id)

        if path.suffix == ".spipe":
            with tempfile.TemporaryDirectory(prefix="spipe-save-") as staging:
                write_manifest(self._manifest, staging)
                if artifacts == "fat" and referenced:
                    self._store.copy_into(Path(staging) / ARTIFACTS_DIR, referenced)
                path.parent.mkdir(parents=True, exist_ok=True)
                pack_zip(staging, path)
            return path

        if self._base_dir is not None and path.exists() \
                and path.resolve() == Path(self._base_dir).resolve():
            # saving onto the backing directory rewrites the manifest in place; a fat save also
            # embeds artifacts the store resolves externally, and a thin save is refused when the
            # directory embeds artifacts since it would have to delete them
            if artifacts == "thin" and (path / ARTIFACTS_DIR).is_dir():
                raise SpipeSaveError(
                    f"A thin export onto the bundle's own directory {path} would delete its "
                    "embedded artifacts; save the thin export to a new path."
                )
            write_manifest(self._manifest, path)
            if artifacts == "fat" and referenced:
                self._store.copy_into(path / ARTIFACTS_DIR, referenced)
            return path

        if path.exists():
            if not path.is_dir():
                raise SpipeSaveError(f"{path} exists and is not a directory.")
            occupied = any(path.iterdir())
            if occupied and not (path / MANIFEST_NAME).exists():
                raise SpipeSaveError(
                    f"{path} exists and is not a spipe directory; refusing to replace it."
                )
            for member in (path / MANIFEST_NAME, path / ARTIFACTS_DIR):
                if member.is_dir():
                    shutil.rmtree(member)
                elif member.exists():
                    member.unlink()
        path.mkdir(parents=True, exist_ok=True)
        write_manifest(self._manifest, path)
        if artifacts == "fat" and referenced:
            self._store.copy_into(path / ARTIFACTS_DIR, referenced)
        return path

    # pipeline construction

    def pipeline(
        self,
        *,
        backend: BackendSpec | str | None = None,
        prefer: str = "frozen",
        verify: str = "strict",
        **pipeline_kwargs,
    ) -> SteeringPipeline:
        """Instantiate a `SteeringPipeline` from this spipe.

        Frozen entries instantiate from their resolution by default; `prefer="recipe"`
        instantiates every entry from its recipe args (forcing re-fits at `steer()`). The
        spipe supplies the model reference and the controls only; backend, device, dtype, and
        `hf_model_kwargs` are the caller's. The caller runs `pipeline.check()` /
        `pipeline.steer()` as normal.

        Args:
            backend: Forwarded to the `SteeringPipeline` constructor.
            prefer: `"frozen"` (default) or `"recipe"`.
            verify: Verification policy for frozen steering artifacts (`"strict"`, `"warn"`,
                or `"off"`), enforced where binding happens at `steer()`.

        Returns:
            The constructed (unsteered) `SteeringPipeline`.

        Raises:
            SpipeFormatError: If a method key resolves to no registered method.
            SpipeCodeRefError: If decoding requires code and the spipe was loaded without
                `allow_code=True`.
            SpipeStaleError: If a deferred staleness check fails (thin bundles).
        """
        if prefer not in ("frozen", "recipe"):
            raise ValueError(f"prefer must be 'frozen' or 'recipe'; got {prefer!r}.")
        if verify not in ("strict", "warn", "off"):
            raise ValueError(f"verify must be 'strict', 'warn', or 'off'; got {verify!r}.")

        if prefer == "frozen" and not self._allow_stale:
            self._check_staleness(raise_on_stale=True)

        ctx = DecodeContext(
            store=self._store,
            allow_code=self._allow_code,
            code_mode="strict",
            verify=verify,
            data_mode="load",
            manifest_records=self._manifest_records,
        )

        controls = []
        for entry in self._manifest["controls"]:
            items = _resolved_items(entry) if prefer == "frozen" else []
            if items:
                for item in items:
                    args = dict(item["args"])
                    if verify != "strict" and item["method"] == "output_control/routed_decoding":
                        args["allow_model_mismatch"] = True
                    if verify != "strict" and item["method"] == "structural_control/load_lora":
                        args["allow_base_mismatch"] = True
                    control = self._instantiate(item["method"], args, ctx)
                    control.enabled = entry["enabled"]
                    controls.append(control)
            else:
                control = self._instantiate(entry["method"], entry["args"], ctx)
                control.enabled = entry["enabled"]
                controls.append(control)

        # controls may hold paths into this spipe's extraction directory; retaining the spipe
        # on each control keeps that directory alive for the pipeline's lifetime
        for control in controls:
            control._spipe_retainer = self

        from steerability.algorithms.core.steering_pipeline import SteeringPipeline

        return SteeringPipeline(
            model_name_or_path=self._manifest["model"]["ref"],
            controls=controls,
            backend=backend,
            **pipeline_kwargs,
        )

code_dependent property

Whether loading this spipe needs matching code and allow_code=True.

config_id property

The configuration identity digest over the recipe entries, 12 hex characters.

is_frozen property

True iff every enabled entry with steer-time products carries a resolution.

Entries with resolved: null have nothing to pin (their recipe is their frozen form), which makes a manifest with a lock section frozen.

manifest property

A deep copy of the manifest.

recipe_id property

The recipe identity digest (model reference plus controls), 12 hex characters.

describe()

A human-readable table of entries, frozen state, and artifact sizes.

Source code in steerability/spipe/spipe.py

def describe(self) -> str:
    """A human-readable table of entries, frozen state, and artifact sizes."""
    lines = [
        f"spipe {self._manifest['format']}  model={self._manifest['model']['ref']}",
        f"  recipe_id={self.recipe_id}  config_id={self.config_id}  "
        f"frozen={self.is_frozen}  code_dependent={self.code_dependent}",
    ]
    for i, entry in enumerate(self._manifest["controls"]):
        items = _resolved_items(entry)
        frozen = "recipe"
        if items:
            methods = ", ".join(item["method"] for item in items)
            frozen = f"frozen -> {methods}"
        elif self.is_frozen and entry["enabled"]:
            frozen = "frozen (recipe is the frozen form)"
        enabled = "" if entry["enabled"] else "  [disabled]"
        lines.append(f"  [{i}] {entry['method']}{enabled}  ({frozen})")
        for item in items:
            for name, record in (item.get("artifacts") or {}).items():
                size = ""
                if self._store is not None and self._store.has(record["id"]):
                    size = f"  {self._store.size_of(record['id'])} bytes"
                lines.append(f"        {name}: {record['type']} {record['id'][:19]}…{size}")
    return "\n".join(lines)

load(path, *, allow_code=False, allow_stale=False, artifact_store=None) classmethod

Load a .spipe file or directory.

Parameters:

Name	Type	Description	Default
path	str | Path	A .spipe zip file or a spipe directory.	required
allow_code	bool	Permit $ref imports, non-steerability. $dc imports, and pickle-backed memory payloads during decoding.	False
allow_stale	bool	Skip the staleness check.	False
artifact_store	str | Path | Callable[[str], Path] | None	External artifact source for thin bundles: a store directory, or a callable mapping an artifact id to the directory holding it.	None

Returns:

Type	Description
'SPipe'	The loaded SPipe.

Raises:

Type	Description
SpipeFormatError	On format-version, schema, or archive violations.
SpipeIntegrityError	If a present artifact fails content verification.
SpipeStaleError	If a frozen entry is stale and allow_stale is False.

Source code in steerability/spipe/spipe.py

@classmethod
def load(
    cls,
    path: str | Path,
    *,
    allow_code: bool = False,
    allow_stale: bool = False,
    artifact_store: str | Path | Callable[[str], Path] | None = None,
) -> "SPipe":
    """Load a `.spipe` file or directory.

    Args:
        path: A `.spipe` zip file or a spipe directory.
        allow_code: Permit `$ref` imports, non-`steerability.` `$dc` imports, and
            pickle-backed memory payloads during decoding.
        allow_stale: Skip the staleness check.
        artifact_store: External artifact source for thin bundles: a store directory, or
            a callable mapping an artifact id to the directory holding it.

    Returns:
        The loaded `SPipe`.

    Raises:
        SpipeFormatError: On format-version, schema, or archive violations.
        SpipeIntegrityError: If a present artifact fails content verification.
        SpipeStaleError: If a frozen entry is stale and `allow_stale` is False.
    """
    path = Path(path)
    temp = None
    if path.is_dir():
        base_dir = path
    else:
        temp = tempfile.TemporaryDirectory(prefix="spipe-load-")
        unpack_zip(path, temp.name)
        base_dir = Path(temp.name)

    manifest = read_manifest(base_dir)

    resolver: Callable[[str], Path] | None = None
    if callable(artifact_store):
        resolver = artifact_store
    elif artifact_store is not None:
        external = Path(artifact_store)
        resolver = lambda artifact_id: external / artifact_id.replace(":", "-", 1)  # noqa: E731
    store = ArtifactStore(base_dir / ARTIFACTS_DIR, resolver=resolver)

    spipe = cls(
        manifest, store=store, base_dir=base_dir,
        allow_code=allow_code, allow_stale=allow_stale, _temp=temp,
    )

    referenced = spipe._referenced_artifact_ids()
    available = [artifact_id for artifact_id in referenced if store.has(artifact_id)]
    for artifact_id in sorted(available):
        store.verify(artifact_id)
    if len(available) < len(referenced):
        logger.info(
            "Thin spipe: %d of %d referenced artifacts unavailable; integrity and "
            "staleness checks defer to pipeline construction.",
            len(referenced) - len(available), len(referenced),
        )
    elif not allow_stale:
        spipe._check_staleness(raise_on_stale=True)

    return spipe

pipeline(*, backend=None, prefer='frozen', verify='strict', **pipeline_kwargs)

Instantiate a SteeringPipeline from this spipe.

Frozen entries instantiate from their resolution by default; prefer="recipe" instantiates every entry from its recipe args (forcing re-fits at steer()). The spipe supplies the model reference and the controls only; backend, device, dtype, and hf_model_kwargs are the caller's. The caller runs pipeline.check() / pipeline.steer() as normal.

Parameters:

Name	Type	Description	Default
backend	BackendSpec | str | None	Forwarded to the SteeringPipeline constructor.	None
prefer	str	"frozen" (default) or "recipe".	'frozen'
verify	str	Verification policy for frozen steering artifacts ("strict", "warn", or "off"), enforced where binding happens at steer().	'strict'

Returns:

Type	Description
SteeringPipeline	The constructed (unsteered) SteeringPipeline.

Raises:

Type	Description
SpipeFormatError	If a method key resolves to no registered method.
SpipeCodeRefError	If decoding requires code and the spipe was loaded without allow_code=True.
SpipeStaleError	If a deferred staleness check fails (thin bundles).

Source code in steerability/spipe/spipe.py

def pipeline(
    self,
    *,
    backend: BackendSpec | str | None = None,
    prefer: str = "frozen",
    verify: str = "strict",
    **pipeline_kwargs,
) -> SteeringPipeline:
    """Instantiate a `SteeringPipeline` from this spipe.

    Frozen entries instantiate from their resolution by default; `prefer="recipe"`
    instantiates every entry from its recipe args (forcing re-fits at `steer()`). The
    spipe supplies the model reference and the controls only; backend, device, dtype, and
    `hf_model_kwargs` are the caller's. The caller runs `pipeline.check()` /
    `pipeline.steer()` as normal.

    Args:
        backend: Forwarded to the `SteeringPipeline` constructor.
        prefer: `"frozen"` (default) or `"recipe"`.
        verify: Verification policy for frozen steering artifacts (`"strict"`, `"warn"`,
            or `"off"`), enforced where binding happens at `steer()`.

    Returns:
        The constructed (unsteered) `SteeringPipeline`.

    Raises:
        SpipeFormatError: If a method key resolves to no registered method.
        SpipeCodeRefError: If decoding requires code and the spipe was loaded without
            `allow_code=True`.
        SpipeStaleError: If a deferred staleness check fails (thin bundles).
    """
    if prefer not in ("frozen", "recipe"):
        raise ValueError(f"prefer must be 'frozen' or 'recipe'; got {prefer!r}.")
    if verify not in ("strict", "warn", "off"):
        raise ValueError(f"verify must be 'strict', 'warn', or 'off'; got {verify!r}.")

    if prefer == "frozen" and not self._allow_stale:
        self._check_staleness(raise_on_stale=True)

    ctx = DecodeContext(
        store=self._store,
        allow_code=self._allow_code,
        code_mode="strict",
        verify=verify,
        data_mode="load",
        manifest_records=self._manifest_records,
    )

    controls = []
    for entry in self._manifest["controls"]:
        items = _resolved_items(entry) if prefer == "frozen" else []
        if items:
            for item in items:
                args = dict(item["args"])
                if verify != "strict" and item["method"] == "output_control/routed_decoding":
                    args["allow_model_mismatch"] = True
                if verify != "strict" and item["method"] == "structural_control/load_lora":
                    args["allow_base_mismatch"] = True
                control = self._instantiate(item["method"], args, ctx)
                control.enabled = entry["enabled"]
                controls.append(control)
        else:
            control = self._instantiate(entry["method"], entry["args"], ctx)
            control.enabled = entry["enabled"]
            controls.append(control)

    # controls may hold paths into this spipe's extraction directory; retaining the spipe
    # on each control keeps that directory alive for the pipeline's lifetime
    for control in controls:
        control._spipe_retainer = self

    from steerability.algorithms.core.steering_pipeline import SteeringPipeline

    return SteeringPipeline(
        model_name_or_path=self._manifest["model"]["ref"],
        controls=controls,
        backend=backend,
        **pipeline_kwargs,
    )

save(path, *, artifacts='fat')

Write the spipe to path.

A path ending in .spipe produces a zip file; any other path produces (or replaces) a directory. artifacts="fat" (default) embeds every referenced artifact; artifacts="thin" writes the manifest only, leaving artifact ids resolvable at load via artifact_store=. A recipe-only spipe with no artifact references writes no artifacts/ directory either way.

Saving onto the bundle's own backing directory rewrites the manifest in place. A fat save there also embeds any referenced artifact the store resolves externally, and a thin save there raises SpipeSaveError when the directory embeds artifacts, since a thin export would have to delete them.

Parameters:

Name	Type	Description	Default
path	str | Path	Destination file or directory.	required
artifacts	str	"fat" or "thin".	'fat'

Returns:

Type	Description
Path	The written path.

Raises:

Type	Description
SpipeSaveError	If a directory target exists and is neither empty nor a spipe directory, a referenced artifact is unavailable for a fat export, or a thin export targets the bundle's own directory while it embeds artifacts.

Source code in steerability/spipe/spipe.py

def save(self, path: str | Path, *, artifacts: str = "fat") -> Path:
    """Write the spipe to `path`.

    A path ending in `.spipe` produces a zip file; any other path produces (or replaces)
    a directory. `artifacts="fat"` (default) embeds every referenced artifact;
    `artifacts="thin"` writes the manifest only, leaving artifact ids resolvable at load
    via `artifact_store=`. A recipe-only spipe with no artifact references writes no
    `artifacts/` directory either way.

    Saving onto the bundle's own backing directory rewrites the manifest in place. A fat
    save there also embeds any referenced artifact the store resolves externally, and a
    thin save there raises `SpipeSaveError` when the directory embeds artifacts, since a
    thin export would have to delete them.

    Args:
        path: Destination file or directory.
        artifacts: `"fat"` or `"thin"`.

    Returns:
        The written path.

    Raises:
        SpipeSaveError: If a directory target exists and is neither empty nor a spipe
            directory, a referenced artifact is unavailable for a fat export, or a thin
            export targets the bundle's own directory while it embeds artifacts.
    """
    if artifacts not in ("fat", "thin"):
        raise SpipeSaveError(f"artifacts must be 'fat' or 'thin'; got {artifacts!r}.")
    path = Path(path)
    referenced = sorted(self._referenced_artifact_ids())
    if artifacts == "fat" and referenced:
        if self._store is None:
            raise SpipeSaveError("This spipe has no artifact store; only artifacts='thin' is possible.")
        for artifact_id in referenced:
            self._store.verify(artifact_id)

    if path.suffix == ".spipe":
        with tempfile.TemporaryDirectory(prefix="spipe-save-") as staging:
            write_manifest(self._manifest, staging)
            if artifacts == "fat" and referenced:
                self._store.copy_into(Path(staging) / ARTIFACTS_DIR, referenced)
            path.parent.mkdir(parents=True, exist_ok=True)
            pack_zip(staging, path)
        return path

    if self._base_dir is not None and path.exists() \
            and path.resolve() == Path(self._base_dir).resolve():
        # saving onto the backing directory rewrites the manifest in place; a fat save also
        # embeds artifacts the store resolves externally, and a thin save is refused when the
        # directory embeds artifacts since it would have to delete them
        if artifacts == "thin" and (path / ARTIFACTS_DIR).is_dir():
            raise SpipeSaveError(
                f"A thin export onto the bundle's own directory {path} would delete its "
                "embedded artifacts; save the thin export to a new path."
            )
        write_manifest(self._manifest, path)
        if artifacts == "fat" and referenced:
            self._store.copy_into(path / ARTIFACTS_DIR, referenced)
        return path

    if path.exists():
        if not path.is_dir():
            raise SpipeSaveError(f"{path} exists and is not a directory.")
        occupied = any(path.iterdir())
        if occupied and not (path / MANIFEST_NAME).exists():
            raise SpipeSaveError(
                f"{path} exists and is not a spipe directory; refusing to replace it."
            )
        for member in (path / MANIFEST_NAME, path / ARTIFACTS_DIR):
            if member.is_dir():
                shutil.rmtree(member)
            elif member.exists():
                member.unlink()
    path.mkdir(parents=True, exist_ok=True)
    write_manifest(self._manifest, path)
    if artifacts == "fat" and referenced:
        self._store.copy_into(path / ARTIFACTS_DIR, referenced)
    return path

thaw()

A recipe-only copy: every resolved section and the lock are dropped.

Source code in steerability/spipe/spipe.py

def thaw(self) -> "SPipe":
    """A recipe-only copy: every `resolved` section and the lock are dropped."""
    manifest = self.manifest
    for entry in manifest["controls"]:
        entry["resolved"] = None
    manifest["lock"] = None
    return SPipe(
        manifest, store=self._store, base_dir=self._base_dir,
        allow_code=self._allow_code, allow_stale=self._allow_stale, _temp=self._temp,
    )

verify()

The model-free report: format, integrity, staleness, versions, code dependence.

Never loads a model or instantiates a backend. Referenced artifacts that are not present are reported in one warning (a thin bundle); the present ones are verified against their content ids.

Returns:

Type	Description
SpipeReport	The report.

Source code in steerability/spipe/spipe.py

def verify(self) -> SpipeReport:
    """The model-free report: format, integrity, staleness, versions, code dependence.

    Never loads a model or instantiates a backend. Referenced artifacts that are not
    present are reported in one warning (a thin bundle); the present ones are verified
    against their content ids.

    Returns:
        The report.
    """
    errors: list[str] = []
    report_warnings: list[str] = []

    try:
        validate_manifest(self._manifest)
    except SpipeFormatError as exc:
        errors.append(str(exc))

    referenced = sorted(self._referenced_artifact_ids())
    missing = [
        artifact_id for artifact_id in referenced
        if self._store is None or not self._store.has(artifact_id)
    ]
    if missing:
        report_warnings.append(
            f"{len(missing)} of {len(referenced)} referenced artifact(s) are not present (thin "
            f"bundle); pass artifact_store= at load to resolve them: {', '.join(missing)}."
        )
    for artifact_id in referenced:
        if artifact_id not in missing:
            try:
                self._store.verify(artifact_id)
            except SpipeIntegrityError as exc:
                errors.append(str(exc))

    try:
        self._check_staleness(raise_on_stale=True, force=True)
    except SpipeStaleError as exc:
        errors.append(str(exc))
    except Exception as exc:
        report_warnings.append(f"staleness check could not run: {exc}")

    from steerability.spipe.freeze import _toolkit_version

    saved = self._manifest.get("toolkit_version", "unknown")
    current = _toolkit_version()
    if saved.split(".")[0] != current.split(".")[0]:
        report_warnings.append(
            f"spipe was written by steerability {saved}; this is {current} (major mismatch)."
        )
    if self.code_dependent:
        report_warnings.append(
            "the manifest references code ($ref); loading a pipeline from it requires "
            "allow_code=True and the referenced modules on the import path."
        )

    return SpipeReport(ok=not errors, errors=errors, warnings=report_warnings)

SpipeCodeRefError

Bases: SpipeError

Decoding requires importing code and allow_code was not granted.

Raised when a manifest contains a $ref callable reference, a $dc class outside the steerability. namespace, or a pickle-backed memory artifact, and load() was called without allow_code=True.

Source code in steerability/spipe/errors.py

class SpipeCodeRefError(SpipeError):
    """Decoding requires importing code and `allow_code` was not granted.

    Raised when a manifest contains a `$ref` callable reference, a `$dc` class outside the
    `steerability.` namespace, or a pickle-backed memory artifact, and `load()` was called
    without `allow_code=True`.
    """

SpipeError

Bases: Exception

Base class for .spipe errors.

Source code in steerability/spipe/errors.py

class SpipeError(Exception):
    """Base class for `.spipe` errors."""

SpipeFormatError

Bases: SpipeError

The manifest or archive violates the spipe/1 format.

Raised for unsupported format versions, schema violations (missing, mistyped, or unknown fields), unknown method keys, and malformed archives.

Source code in steerability/spipe/errors.py

class SpipeFormatError(SpipeError):
    """The manifest or archive violates the `spipe/1` format.

    Raised for unsupported format versions, schema violations (missing, mistyped, or unknown
    fields), unknown method keys, and malformed archives.
    """

SpipeIntegrityError

Bases: SpipeError

Stored artifact bytes do not match their content-addressed id.

Source code in steerability/spipe/errors.py

class SpipeIntegrityError(SpipeError):
    """Stored artifact bytes do not match their content-addressed id."""

SpipeReport dataclass

The result of a model-free SPipe.verify().

Attributes:

Name	Type	Description
ok	bool	True when no errors were found.
errors	list[str]	Findings that make the bundle unusable as-is.
warnings	list[str]	Findings worth knowing that do not block loading.

Source code in steerability/spipe/spipe.py

@dataclass
class SpipeReport:
    """The result of a model-free `SPipe.verify()`.

    Attributes:
        ok: True when no errors were found.
        errors: Findings that make the bundle unusable as-is.
        warnings: Findings worth knowing that do not block loading.
    """

    ok: bool
    errors: list[str] = field(default_factory=list)
    warnings: list[str] = field(default_factory=list)

    def render(self) -> str:
        """A human-readable summary."""
        lines = [f"spipe verify: {'ok' if self.ok else 'FAILED'}"]
        lines.extend(f"  error: {message}" for message in self.errors)
        lines.extend(f"  warning: {message}" for message in self.warnings)
        return "\n".join(lines)

    def __str__(self) -> str:
        return self.render()

errors = field(default_factory=list) class-attribute instance-attribute

ok instance-attribute

warnings = field(default_factory=list) class-attribute instance-attribute

render()

A human-readable summary.

Source code in steerability/spipe/spipe.py

def render(self) -> str:
    """A human-readable summary."""
    lines = [f"spipe verify: {'ok' if self.ok else 'FAILED'}"]
    lines.extend(f"  error: {message}" for message in self.errors)
    lines.extend(f"  warning: {message}" for message in self.warnings)
    return "\n".join(lines)

SpipeSaveError

Bases: SpipeError

A pipeline or value cannot be serialized.

Raised for unresolvable model references, unregistered control classes, live model or tokenizer objects inside args, lambdas and other unnameable callables, reserved $-prefixed mapping keys, values over the inline size limit, and controls that cannot freeze.

Source code in steerability/spipe/errors.py

class SpipeSaveError(SpipeError):
    """A pipeline or value cannot be serialized.

    Raised for unresolvable model references, unregistered control classes, live model or
    tokenizer objects inside args, lambdas and other unnameable callables, reserved `$`-prefixed
    mapping keys, values over the inline size limit, and controls that cannot freeze.
    """

SpipeStaleError

Bases: SpipeError

A frozen artifact's recorded fit digest does not match the current recipe.

The recipe's fit-relevant fields were edited after freezing and the pinned artifacts no longer correspond to the recipe. Call thaw() and re-steer(), or pass allow_stale=True to load anyway.

Source code in steerability/spipe/errors.py

class SpipeStaleError(SpipeError):
    """A frozen artifact's recorded fit digest does not match the current recipe.

    The recipe's fit-relevant fields were edited after freezing and the pinned artifacts no
    longer correspond to the recipe. Call `thaw()` and re-`steer()`, or pass
    `allow_stale=True` to load anyway.
    """

codec

The .spipe value codec: round-tripping between constructor values and manifest JSON.

Plain JSON values pass through; everything else encodes as a single-key tagged object:

• $artifact: tensor-bearing or on-disk state, stored content-addressed (store.py).
• $dc: pure-data dataclasses and enums, decoded by importing the class and calling it.
• $component: state-control components (transforms, gates, selectors) and routing objects, decoded through COMPONENT_KINDS.
• $ref: module-level callables, decoded by import only under allow_code=True.
• $data: dataset references (DataRef), materialized at pipeline construction.
• $map: mappings with non-string scalar keys, kept as [key, value] pairs sorted by key so key types are retained through the JSON round trip and the encoding is independent of insertion order.

The codec round-trips; identity.canonical_value remains the separate, hash-only canonicalizer. Digest computation for fit identities goes through encode in digest mode (EncodeContext(store=None)), where tensors reduce to content ids without being written and unhandled objects reduce to their type name.

COMPONENT_KINDS = _component_kinds module-attribute

TAGS = ('$artifact', '$dc', '$component', '$ref', '$data', '$map') module-attribute

TRUSTED_DC_PREFIX = 'steerability.' module-attribute

logger = logging.getLogger(__name__) module-attribute

AsPath

Marker wrapping an artifact value whose decoded form is the payload path.

Used by frozen_form implementations whose constructor field takes a path rather than the reconstructed object (e.g. sasa's wv_path, load_lora's path).

Source code in steerability/spipe/codec.py

class AsPath:
    """Marker wrapping an artifact value whose decoded form is the payload path.

    Used by `frozen_form` implementations whose constructor field takes a path rather than the
    reconstructed object (e.g. `sasa`'s `wv_path`, `load_lora`'s `path`).
    """

    def __init__(self, value: Any):
        self.value = value

value = value instance-attribute

CodeRef dataclass

Inert stand-in for a $ref callable, used by internal digest decoding.

Encodes back to the same $ref; calling it raises.

Attributes:

Name	Type	Description
target	str	The "<module>:<qualname>" reference.

Source code in steerability/spipe/codec.py

@dataclass(frozen=True)
class CodeRef:
    """Inert stand-in for a `$ref` callable, used by internal digest decoding.

    Encodes back to the same `$ref`; calling it raises.

    Attributes:
        target: The `"<module>:<qualname>"` reference.
    """

    target: str

    def __call__(self, *args, **kwargs):
        raise SpipeCodeRefError(
            f"Callable reference {self.target!r} was not resolved; load the spipe with "
            "allow_code=True to import it."
        )

target instance-attribute

DataRef dataclass

A reference to a dataset, resolved at pipeline construction.

Attributes:

Name	Type	Description
kind	str	"hf" (a Hugging Face Hub dataset), "path" (a digest-pinned local file), or "opaque" (an in-memory dataset recorded by fingerprint only, not reloadable).
repo_id	str | None	Hub dataset id ("hf" only).
revision	str | None	Hub revision ("hf" only).
split	str | None	Dataset split ("hf" only).
subset	str | None	Dataset configuration name ("hf" only).
path	str | None	Local file path ("path" only).
sha256	str | None	Hex digest the file must match ("path" only).
type	str | None	Type qualname of the recorded object ("opaque" only).
fingerprint	str | None	Content fingerprint of the recorded object ("opaque" only), or None.

Source code in steerability/spipe/codec.py

@dataclass(frozen=True)
class DataRef:
    """A reference to a dataset, resolved at pipeline construction.

    Attributes:
        kind: `"hf"` (a Hugging Face Hub dataset), `"path"` (a digest-pinned local file), or
            `"opaque"` (an in-memory dataset recorded by fingerprint only, not reloadable).
        repo_id: Hub dataset id (`"hf"` only).
        revision: Hub revision (`"hf"` only).
        split: Dataset split (`"hf"` only).
        subset: Dataset configuration name (`"hf"` only).
        path: Local file path (`"path"` only).
        sha256: Hex digest the file must match (`"path"` only).
        type: Type qualname of the recorded object (`"opaque"` only).
        fingerprint: Content fingerprint of the recorded object (`"opaque"` only), or None.
    """

    kind: str
    repo_id: str | None = None
    revision: str | None = None
    split: str | None = None
    subset: str | None = None
    path: str | None = None
    sha256: str | None = None
    type: str | None = None
    fingerprint: str | None = None

    def load(self) -> Any:
        """Materialize the reference.

        Returns:
            The loaded dataset for `"hf"`, the verified path string for `"path"`.

        Raises:
            SpipeFormatError: If the kind is `"opaque"` (not reloadable) or unknown.
            SpipeIntegrityError: If a `"path"` file does not match its recorded digest.
        """
        if self.kind == "hf":
            from datasets import load_dataset

            kwargs: dict[str, Any] = {}
            if self.revision is not None:
                kwargs["revision"] = self.revision
            if self.split is not None:
                kwargs["split"] = self.split
            if self.subset is not None:
                return load_dataset(self.repo_id, self.subset, **kwargs)
            return load_dataset(self.repo_id, **kwargs)
        if self.kind == "path":
            import hashlib

            from steerability.spipe.errors import SpipeIntegrityError

            digest = hashlib.sha256(Path(self.path).read_bytes()).hexdigest()
            if self.sha256 is not None and digest != self.sha256:
                raise SpipeIntegrityError(
                    f"Data file {self.path} hashes to {digest}, expected {self.sha256}."
                )
            return self.path
        raise SpipeFormatError(
            f"DataRef kind {self.kind!r} cannot be materialized; an opaque dataset reference "
            "records identity only. Re-supply the dataset (or a 'hf'/'path' DataRef) to re-steer."
        )

fingerprint = None class-attribute instance-attribute

kind instance-attribute

path = None class-attribute instance-attribute

repo_id = None class-attribute instance-attribute

revision = None class-attribute instance-attribute

sha256 = None class-attribute instance-attribute

split = None class-attribute instance-attribute

subset = None class-attribute instance-attribute

type = None class-attribute instance-attribute

load()

Materialize the reference.

Returns:

Type	Description
Any	The loaded dataset for "hf", the verified path string for "path".

Raises:

Type	Description
SpipeFormatError	If the kind is "opaque" (not reloadable) or unknown.
SpipeIntegrityError	If a "path" file does not match its recorded digest.

Source code in steerability/spipe/codec.py

def load(self) -> Any:
    """Materialize the reference.

    Returns:
        The loaded dataset for `"hf"`, the verified path string for `"path"`.

    Raises:
        SpipeFormatError: If the kind is `"opaque"` (not reloadable) or unknown.
        SpipeIntegrityError: If a `"path"` file does not match its recorded digest.
    """
    if self.kind == "hf":
        from datasets import load_dataset

        kwargs: dict[str, Any] = {}
        if self.revision is not None:
            kwargs["revision"] = self.revision
        if self.split is not None:
            kwargs["split"] = self.split
        if self.subset is not None:
            return load_dataset(self.repo_id, self.subset, **kwargs)
        return load_dataset(self.repo_id, **kwargs)
    if self.kind == "path":
        import hashlib

        from steerability.spipe.errors import SpipeIntegrityError

        digest = hashlib.sha256(Path(self.path).read_bytes()).hexdigest()
        if self.sha256 is not None and digest != self.sha256:
            raise SpipeIntegrityError(
                f"Data file {self.path} hashes to {digest}, expected {self.sha256}."
            )
        return self.path
    raise SpipeFormatError(
        f"DataRef kind {self.kind!r} cannot be materialized; an opaque dataset reference "
        "records identity only. Re-supply the dataset (or a 'hf'/'path' DataRef) to re-steer."
    )

DecodeContext dataclass

State threaded through one decoding pass.

Attributes:

Name	Type	Description
store	ArtifactStore | None	Source artifact store, or None when artifacts are unavailable (thin bundle without an external store).
allow_code	bool	Whether $ref imports, non-steerability. $dc imports, and pickle-backed memory payloads are permitted.
code_mode	str	"strict" raises on ungranted code references; "sentinel" substitutes inert CodeRef markers for $ref without importing, even under allow_code (internal digest decoding only).
verify	str	Verification policy applied to frozen steering artifacts ("strict", "warn", or "off").
data_mode	str	"load" materializes $data references; "keep" returns DataRef values unchanged.
manifest_records	Mapping[str, ArtifactRecord]	Artifact records from the manifest's resolved entries, keyed by artifact id. When an artifact has one, its artifact_class and provenance decide the verification wrap; the store sidecar supplies the encoding, type, and reconstruction metadata.

Source code in steerability/spipe/codec.py

@dataclass
class DecodeContext:
    """State threaded through one decoding pass.

    Attributes:
        store: Source artifact store, or None when artifacts are unavailable (thin bundle
            without an external store).
        allow_code: Whether `$ref` imports, non-`steerability.` `$dc` imports, and pickle-backed
            memory payloads are permitted.
        code_mode: `"strict"` raises on ungranted code references; `"sentinel"` substitutes
            inert `CodeRef` markers for `$ref` without importing, even under `allow_code`
            (internal digest decoding only).
        verify: Verification policy applied to frozen steering artifacts (`"strict"`,
            `"warn"`, or `"off"`).
        data_mode: `"load"` materializes `$data` references; `"keep"` returns `DataRef`
            values unchanged.
        manifest_records: Artifact records from the manifest's resolved entries, keyed by
            artifact id. When an artifact has one, its `artifact_class` and `provenance` decide
            the verification wrap; the store sidecar supplies the encoding, type, and
            reconstruction metadata.
    """

    store: ArtifactStore | None = None
    allow_code: bool = False
    code_mode: str = "strict"
    verify: str = "strict"
    data_mode: str = "load"
    manifest_records: Mapping[str, ArtifactRecord] = field(default_factory=dict)

allow_code = False class-attribute instance-attribute

code_mode = 'strict' class-attribute instance-attribute

data_mode = 'load' class-attribute instance-attribute

manifest_records = field(default_factory=dict) class-attribute instance-attribute

store = None class-attribute instance-attribute

verify = 'strict' class-attribute instance-attribute

EncodeContext dataclass

State threaded through one encoding pass.

Attributes:

Name	Type	Description
store	ArtifactStore | None	Destination artifact store, or None for digest mode (tensor ids are computed without writing, unhandled objects reduce to their type name, and callables never raise).
records	dict[str, ArtifactRecord]	Artifact records produced during the pass, keyed by artifact id.
code_refs	list[str]	$ref targets produced during the pass (drives code_dependent).
provenance	dict	Provenance mapping stamped onto artifact records.
default_model_type	str | None	model_type stamped onto exported vectors recorded as "unknown", when the producing model is known.

Source code in steerability/spipe/codec.py

@dataclass
class EncodeContext:
    """State threaded through one encoding pass.

    Attributes:
        store: Destination artifact store, or None for digest mode (tensor ids are computed
            without writing, unhandled objects reduce to their type name, and callables never
            raise).
        records: Artifact records produced during the pass, keyed by artifact id.
        code_refs: `$ref` targets produced during the pass (drives `code_dependent`).
        provenance: Provenance mapping stamped onto artifact records.
        default_model_type: `model_type` stamped onto exported vectors recorded as
            `"unknown"`, when the producing model is known.
    """

    store: ArtifactStore | None = None
    records: dict[str, ArtifactRecord] = field(default_factory=dict)
    code_refs: list[str] = field(default_factory=list)
    provenance: dict = field(default_factory=dict)
    default_model_type: str | None = None

    # per-artifact fields installed by the freeze layer before encoding one state value
    artifact_fields: dict = field(default_factory=dict)

    @property
    def digest_mode(self) -> bool:
        return self.store is None

artifact_fields = field(default_factory=dict) class-attribute instance-attribute

code_refs = field(default_factory=list) class-attribute instance-attribute

default_model_type = None class-attribute instance-attribute

digest_mode property

provenance = field(default_factory=dict) class-attribute instance-attribute

records = field(default_factory=dict) class-attribute instance-attribute

store = None class-attribute instance-attribute

decode(value, ctx, path='$')

Decode one manifest JSON value back to its constructor form.

Parameters:

Name	Type	Description	Default
value	Any	The encoded value.	required
ctx	DecodeContext	The decoding context (store, code and data policies, verification policy).	required
path	str	Breadcrumb naming the position of value, used in error messages.	'$'

Returns:

Type	Description
Any	The decoded value.

Raises:

Type	Description
SpipeFormatError	If a tagged object is malformed or names an unknown kind or type.
SpipeCodeRefError	If decoding requires code and allow_code was not granted.
SpipeIntegrityError	If a referenced artifact is unavailable or fails verification.

Source code in steerability/spipe/codec.py

def decode(value: Any, ctx: DecodeContext, path: str = "$") -> Any:
    """Decode one manifest JSON value back to its constructor form.

    Args:
        value: The encoded value.
        ctx: The decoding context (store, code and data policies, verification policy).
        path: Breadcrumb naming the position of `value`, used in error messages.

    Returns:
        The decoded value.

    Raises:
        SpipeFormatError: If a tagged object is malformed or names an unknown kind or type.
        SpipeCodeRefError: If decoding requires code and `allow_code` was not granted.
        SpipeIntegrityError: If a referenced artifact is unavailable or fails verification.
    """
    if value is None or isinstance(value, (bool, int, float, str)):
        return value
    if isinstance(value, list):
        return [decode(item, ctx, f"{path}[{i}]") for i, item in enumerate(value)]
    if not isinstance(value, Mapping):
        raise SpipeFormatError(f"{path}: unexpected value of type {type(value).__name__}.")

    if "$artifact" in value:
        return _decode_artifact(value, ctx, path)
    if "$map" in value:
        return {
            entry[0]: decode(entry[1], ctx, f"{path}[{entry[0]!r}]")
            for entry in value["$map"]
        }
    if "$dc" in value:
        target = value["$dc"]
        if target == "torch.dtype":
            dtype = getattr(torch, value["value"], None)
            if not isinstance(dtype, torch.dtype):
                raise SpipeFormatError(f"{path}: {value['value']!r} does not name a torch dtype.")
            return dtype
        cls = _import_qualname(target, ctx, path)
        if isinstance(cls, type) and issubclass(cls, Enum):
            return cls[value["value"]]
        decoded_fields = {
            name: decode(item, ctx, f"{path}.{name}")
            for name, item in value.get("fields", {}).items()
        }
        return cls(**decoded_fields)
    if "$component" in value:
        return _decode_component(value, ctx, path)
    if "$ref" in value:
        target = value["$ref"]
        if ctx.code_mode == "sentinel":
            return CodeRef(target)
        if ctx.allow_code:
            module_name, _, qual = target.partition(":")
            obj: Any = import_module(module_name)
            for part in qual.split("."):
                obj = getattr(obj, part)
            return obj
        raise SpipeCodeRefError(
            f"{path}: this spipe references code ({target!r}); pass allow_code=True to load() "
            "to import it."
        )
    if "$data" in value:
        payload = value["$data"]
        ref = DataRef(
            kind=payload.get("kind", "opaque"),
            repo_id=payload.get("repo_id"),
            revision=payload.get("revision"),
            split=payload.get("split"),
            subset=payload.get("subset"),
            path=payload.get("path"),
            sha256=payload.get("sha256"),
            type=payload.get("type"),
            fingerprint=payload.get("fingerprint"),
        )
        if ctx.data_mode == "load" and ref.kind in ("hf", "path"):
            return ref.load()
        return ref

    return {key: decode(item, ctx, f"{path}[{key!r}]") for key, item in value.items()}

digest_of(value)

12-hex-character digest of a value's encoded form, in digest mode.

Tensors reduce to content-addressed ids (float32 form), callables to their qualified names, and unhandled objects to their type names, which keeps the digest stable across a save and load round trip.

Parameters:

Name	Type	Description	Default
value	Any	The value to digest.	required

Returns:

Type	Description
str	The digest.

Source code in steerability/spipe/codec.py

def digest_of(value: Any) -> str:
    """12-hex-character digest of a value's encoded form, in digest mode.

    Tensors reduce to content-addressed ids (float32 form), callables to their qualified
    names, and unhandled objects to their type names, which keeps the digest stable across a
    save and load round trip.

    Args:
        value: The value to digest.

    Returns:
        The digest.
    """
    import hashlib

    encoded = encode(value, EncodeContext(store=None))
    serialized = json.dumps(encoded, sort_keys=True)
    return hashlib.sha256(serialized.encode("utf-8")).hexdigest()[:12]

encode(value, ctx, path='$')

Encode one constructor value to its manifest JSON form.

Parameters:

Name	Type	Description	Default
value	Any	The value to encode.	required
ctx	EncodeContext	The encoding context; artifacts land in ctx.store (or reduce to ids in digest mode) and $ref targets accumulate on ctx.code_refs.	required
path	str	Breadcrumb naming the position of value, used in error messages.	'$'

Returns:

Type	Description
Any	A JSON-serializable form of value.

Raises:

Type	Description
SpipeSaveError	If value (or a nested value) has no serialized form; the message names the path and, where one exists, the alternative.

Source code in steerability/spipe/codec.py

def encode(value: Any, ctx: EncodeContext, path: str = "$") -> Any:
    """Encode one constructor value to its manifest JSON form.

    Args:
        value: The value to encode.
        ctx: The encoding context; artifacts land in `ctx.store` (or reduce to ids in digest
            mode) and `$ref` targets accumulate on `ctx.code_refs`.
        path: Breadcrumb naming the position of `value`, used in error messages.

    Returns:
        A JSON-serializable form of `value`.

    Raises:
        SpipeSaveError: If `value` (or a nested value) has no serialized form; the message
            names the path and, where one exists, the alternative.
    """
    if value is None or isinstance(value, (bool, int, str)):
        return value
    if isinstance(value, float):
        return value
    if isinstance(value, Path):
        return str(value)
    if isinstance(value, np.generic):
        return value.item()
    if isinstance(value, np.ndarray):
        return value.tolist()
    if isinstance(value, AsPath):
        encoded = _encode_artifact_object(value.value, ctx, path, as_path=True)
        if encoded is None:
            raise SpipeSaveError(f"{path}: AsPath wraps a non-artifact value ({type(value.value).__name__}).")
        return encoded
    if isinstance(value, DataRef):
        payload = {k: v for k, v in (
            ("kind", value.kind), ("repo_id", value.repo_id), ("revision", value.revision),
            ("split", value.split), ("subset", value.subset), ("path", value.path),
            ("sha256", value.sha256), ("type", value.type), ("fingerprint", value.fingerprint),
        ) if v is not None}
        return {"$data": payload}
    if isinstance(value, CodeRef):
        ctx.code_refs.append(value.target)
        return {"$ref": value.target}
    if isinstance(value, torch.dtype):
        return {"$dc": "torch.dtype", "value": str(value).removeprefix("torch.")}
    if isinstance(value, Enum):
        return {"$dc": _qualname(type(value)), "value": value.name}

    artifact = _encode_artifact_object(value, ctx, path)
    if artifact is not None:
        return artifact

    component = _encode_component(value, ctx, path)
    if component is not None:
        return component

    if _looks_like_hf_dataset(value):
        return {"$data": {
            "kind": "opaque",
            "type": _qualname(type(value)),
            "fingerprint": getattr(value, "_fingerprint", None),
        }}

    if is_dataclass(value) and not isinstance(value, type):
        return _encode_dataclass(value, ctx, path)

    if isinstance(value, Mapping):
        if all(isinstance(key, str) for key in value):
            encoded = {}
            for key, item in value.items():
                if key.startswith("$"):
                    raise SpipeSaveError(f"{path}: mapping key {key!r} uses the reserved '$' prefix.")
                encoded[key] = encode(item, ctx, f"{path}[{key!r}]")
            return encoded
        # non-string scalar keys keep their exact type through the $map form
        entries = []
        for key, item in value.items():
            if not isinstance(key, (str, int, float, bool)):
                raise SpipeSaveError(
                    f"{path}: mapping key {key!r} of type {type(key).__name__} has no "
                    "serialized form; mapping keys must be strings, ints, floats, or bools."
                )
            entries.append([key, encode(item, ctx, f"{path}[{key!r}]")])
        entries.sort(key=lambda entry: (type(entry[0]).__name__, repr(entry[0])))
        return {"$map": entries}
    if isinstance(value, (list, tuple)):
        return [encode(item, ctx, f"{path}[{i}]") for i, item in enumerate(value)]
    if isinstance(value, (set, frozenset)):
        encoded_items = [encode(item, ctx, path) for item in value]
        return sorted(encoded_items, key=lambda item: json.dumps(item, sort_keys=True))

    if _is_live_model_or_tokenizer(value):
        raise SpipeSaveError(
            f"{path}: live model/tokenizer objects do not serialize; pass the corresponding "
            "*_name_or_path reference instead."
        )

    if callable(value):
        if ctx.digest_mode:
            try:
                target = _callable_target(value, path)
            except SpipeSaveError:
                target = f"callable:{getattr(value, '__qualname__', type(value).__name__)}"
            return {"$ref": target}
        target = _callable_target(value, path)
        ctx.code_refs.append(target)
        return {"$ref": target}

    if ctx.digest_mode:
        return {"$type": _qualname(type(value))}
    raise SpipeSaveError(
        f"{path}: values of type {_qualname(type(value))} have no serialized form."
    )

encoded_size(encoded)

Byte length of the canonical JSON form of an encoded value.

Source code in steerability/spipe/codec.py

def encoded_size(encoded: Any) -> int:
    """Byte length of the canonical JSON form of an encoded value."""
    return len(json.dumps(encoded, sort_keys=True).encode("utf-8"))

errors

Exception types for the .spipe serialization format.

NotFreezableError

Bases: RuntimeError

A control produces steer-time state but declares no frozen form.

Raised by BaseControl.frozen_form when a control's steer step produces fits or exported state and the control does not override frozen_form to name a constructor-valid frozen form. Re-exported from steerability.spipe.errors alongside the other spipe exceptions.

Source code in steerability/algorithms/core/base_control.py

class NotFreezableError(RuntimeError):
    """A control produces steer-time state but declares no frozen form.

    Raised by `BaseControl.frozen_form` when a control's steer step produces fits or exported
    state and the control does not override `frozen_form` to name a constructor-valid frozen
    form. Re-exported from `steerability.spipe.errors` alongside the other spipe exceptions.
    """

SpipeCodeRefError

Bases: SpipeError

Decoding requires importing code and allow_code was not granted.

Raised when a manifest contains a $ref callable reference, a $dc class outside the steerability. namespace, or a pickle-backed memory artifact, and load() was called without allow_code=True.

Source code in steerability/spipe/errors.py

class SpipeCodeRefError(SpipeError):
    """Decoding requires importing code and `allow_code` was not granted.

    Raised when a manifest contains a `$ref` callable reference, a `$dc` class outside the
    `steerability.` namespace, or a pickle-backed memory artifact, and `load()` was called
    without `allow_code=True`.
    """

SpipeError

Bases: Exception

Base class for .spipe errors.

Source code in steerability/spipe/errors.py

class SpipeError(Exception):
    """Base class for `.spipe` errors."""

SpipeFormatError

Bases: SpipeError

The manifest or archive violates the spipe/1 format.

Raised for unsupported format versions, schema violations (missing, mistyped, or unknown fields), unknown method keys, and malformed archives.

Source code in steerability/spipe/errors.py

class SpipeFormatError(SpipeError):
    """The manifest or archive violates the `spipe/1` format.

    Raised for unsupported format versions, schema violations (missing, mistyped, or unknown
    fields), unknown method keys, and malformed archives.
    """

SpipeIntegrityError

Bases: SpipeError

Stored artifact bytes do not match their content-addressed id.

Source code in steerability/spipe/errors.py

class SpipeIntegrityError(SpipeError):
    """Stored artifact bytes do not match their content-addressed id."""

SpipeSaveError

Bases: SpipeError

A pipeline or value cannot be serialized.

Raised for unresolvable model references, unregistered control classes, live model or tokenizer objects inside args, lambdas and other unnameable callables, reserved $-prefixed mapping keys, values over the inline size limit, and controls that cannot freeze.

Source code in steerability/spipe/errors.py

class SpipeSaveError(SpipeError):
    """A pipeline or value cannot be serialized.

    Raised for unresolvable model references, unregistered control classes, live model or
    tokenizer objects inside args, lambdas and other unnameable callables, reserved `$`-prefixed
    mapping keys, values over the inline size limit, and controls that cannot freeze.
    """

SpipeStaleError

Bases: SpipeError

A frozen artifact's recorded fit digest does not match the current recipe.

The recipe's fit-relevant fields were edited after freezing and the pinned artifacts no longer correspond to the recipe. Call thaw() and re-steer(), or pass allow_stale=True to load anyway.

Source code in steerability/spipe/errors.py

class SpipeStaleError(SpipeError):
    """A frozen artifact's recorded fit digest does not match the current recipe.

    The recipe's fit-relevant fields were edited after freezing and the pinned artifacts no
    longer correspond to the recipe. Call `thaw()` and re-`steer()`, or pass
    `allow_stale=True` to load anyway.
    """

format

The spipe/1 manifest schema and archive packing.

The canonical form of a .spipe is a directory holding spipe.json and an optional artifacts/ store; a .spipe file is a zip of that directory. Zips are written deterministically (sorted member names, stored uncompressed, fixed timestamps) and unpacked with a zip-slip guard, symlink rejection, and a decompressed-size cap.

ARTIFACTS_DIR = 'artifacts' module-attribute

FORMAT_VERSION = 'spipe/1' module-attribute

MANIFEST_NAME = 'spipe.json' module-attribute

SIZE_CAP_BYTES = 20 * 1024 ** 3 module-attribute

logger = logging.getLogger(__name__) module-attribute

pack_zip(directory, target)

Write directory as a deterministic zip at target.

Members are written in sorted arcname order, stored uncompressed, with a fixed (1980, 1, 1, 0, 0, 0) timestamp and no extra fields. Equal directories therefore produce byte-equal files.

Raises:

Type	Description
SpipeFormatError	If directory contains a symlink.

Source code in steerability/spipe/format.py

def pack_zip(directory: str | Path, target: str | Path) -> None:
    """Write `directory` as a deterministic zip at `target`.

    Members are written in sorted arcname order, stored uncompressed, with a fixed
    `(1980, 1, 1, 0, 0, 0)` timestamp and no extra fields. Equal directories therefore
    produce byte-equal files.

    Raises:
        SpipeFormatError: If `directory` contains a symlink.
    """
    directory = Path(directory)
    members = sorted(
        (path for path in directory.rglob("*") if path.is_file() or path.is_symlink()),
        key=lambda path: path.relative_to(directory).as_posix(),
    )
    with zipfile.ZipFile(target, "w", compression=zipfile.ZIP_STORED) as archive:
        for path in members:
            if path.is_symlink():
                raise SpipeFormatError(f"Refusing to pack symlink {path} into a spipe archive.")
            arcname = path.relative_to(directory).as_posix()
            info = zipfile.ZipInfo(arcname, date_time=(1980, 1, 1, 0, 0, 0))
            info.compress_type = zipfile.ZIP_STORED
            info.external_attr = 0o644 << 16
            archive.writestr(info, path.read_bytes())

read_manifest(directory)

Read and validate spipe.json from directory.

Raises:

Type	Description
SpipeFormatError	If the manifest is absent, is not valid JSON, or violates the schema.

Source code in steerability/spipe/format.py

def read_manifest(directory: str | Path) -> dict:
    """Read and validate `spipe.json` from `directory`.

    Raises:
        SpipeFormatError: If the manifest is absent, is not valid JSON, or violates the
            schema.
    """
    path = Path(directory) / MANIFEST_NAME
    if not path.exists():
        raise SpipeFormatError(f"{directory} contains no {MANIFEST_NAME}; not a spipe bundle.")
    try:
        manifest = json.loads(path.read_text(encoding="utf-8"))
    except json.JSONDecodeError as exc:
        raise SpipeFormatError(f"{path} is not valid JSON: {exc}.") from exc
    validate_manifest(manifest)
    return manifest

unpack_zip(archive_path, dest)

Extract a .spipe zip into dest with the format's safety guards.

Every member name is normalized and rejected if absolute, containing .., or resolving outside dest; symlink members are rejected; the total decompressed size is capped at 20 GB.

Raises:

Type	Description
SpipeFormatError	If the archive is not a zip, a member violates the guards, or the size cap is exceeded.

Source code in steerability/spipe/format.py

def unpack_zip(archive_path: str | Path, dest: str | Path) -> None:
    """Extract a `.spipe` zip into `dest` with the format's safety guards.

    Every member name is normalized and rejected if absolute, containing `..`, or resolving
    outside `dest`; symlink members are rejected; the total decompressed size is capped at
    20 GB.

    Raises:
        SpipeFormatError: If the archive is not a zip, a member violates the guards, or the
            size cap is exceeded.
    """
    dest = Path(dest).resolve()
    if not zipfile.is_zipfile(archive_path):
        raise SpipeFormatError(f"{archive_path} is not a zip archive.")
    with zipfile.ZipFile(archive_path) as archive:
        total = sum(info.file_size for info in archive.infolist())
        if total > SIZE_CAP_BYTES:
            raise SpipeFormatError(
                f"Archive decompresses to {total} bytes, over the {SIZE_CAP_BYTES}-byte cap."
            )
        for info in archive.infolist():
            name = info.filename
            if name.startswith(("/", "\\")) or ".." in Path(name).parts or (len(name) > 1 and name[1] == ":"):
                raise SpipeFormatError(f"Archive member {name!r} escapes the extraction root.")
            mode = (info.external_attr >> 16) & 0o170000
            if mode == 0o120000:
                raise SpipeFormatError(f"Archive member {name!r} is a symlink; symlinks are rejected.")
            target = (dest / name).resolve()
            if not target.is_relative_to(dest):
                raise SpipeFormatError(f"Archive member {name!r} escapes the extraction root.")
        for info in archive.infolist():
            if info.is_dir():
                (dest / info.filename).mkdir(parents=True, exist_ok=True)
                continue
            target = dest / info.filename
            target.parent.mkdir(parents=True, exist_ok=True)
            with archive.open(info) as source, open(target, "wb") as sink:
                while chunk := source.read(1 << 20):
                    sink.write(chunk)

validate_manifest(manifest)

Validate a parsed spipe.json against the spipe/1 schema.

Parameters:

Name	Type	Description	Default
manifest	Any	The parsed manifest.	required

Raises:

Type	Description
SpipeFormatError	On any schema violation; the message names the offending field. An unsupported format value is reported first, naming the found and supported versions.

Source code in steerability/spipe/format.py

def validate_manifest(manifest: Any) -> None:
    """Validate a parsed `spipe.json` against the `spipe/1` schema.

    Args:
        manifest: The parsed manifest.

    Raises:
        SpipeFormatError: On any schema violation; the message names the offending field.
            An unsupported `format` value is reported first, naming the found and supported
            versions.
    """
    _require(isinstance(manifest, Mapping), "spipe.json: manifest must be a JSON object.")
    found = manifest.get("format")
    _require(
        found == FORMAT_VERSION,
        f"Unsupported spipe format {found!r}; this toolkit supports {FORMAT_VERSION!r}.",
    )
    _check_keys(manifest, _TOP_LEVEL_KEYS, "spipe.json")
    for key in ("created_at", "toolkit_version"):
        _require(isinstance(manifest.get(key), str), f"spipe.json.{key}: required string.")
    _require(isinstance(manifest.get("code_dependent"), bool), "spipe.json.code_dependent: required boolean.")

    model = manifest.get("model")
    _require(isinstance(model, Mapping), "spipe.json.model: required object.")
    _check_keys(model, _MODEL_KEYS, "spipe.json.model")
    _require(isinstance(model.get("ref"), str) and model["ref"], "spipe.json.model.ref: required non-empty string.")
    _require(model.get("revision") is None or isinstance(model["revision"], str),
             "spipe.json.model.revision: must be a string or null.")

    controls = manifest.get("controls")
    _require(isinstance(controls, list), "spipe.json.controls: required array.")
    for i, entry in enumerate(controls):
        where = f"spipe.json.controls[{i}]"
        _require(isinstance(entry, Mapping), f"{where}: must be an object.")
        _check_keys(entry, _ENTRY_KEYS, where)
        _require(isinstance(entry.get("method"), str) and "/" in entry.get("method", ""),
                 f"{where}.method: required '<category>_control/<name>' string.")
        _require(isinstance(entry.get("enabled"), bool), f"{where}.enabled: required boolean.")
        _require(isinstance(entry.get("args"), Mapping), f"{where}.args: required object.")
        resolved = entry.get("resolved")
        if resolved is None:
            continue
        if isinstance(resolved, list):
            _require(bool(resolved), f"{where}.resolved: an array of frozen entries must be non-empty.")
            for j, item in enumerate(resolved):
                _validate_resolved_entry(item, f"{where}.resolved[{j}]")
        else:
            _validate_resolved_entry(resolved, f"{where}.resolved")

    lock = manifest.get("lock")
    if lock is not None:
        _require(isinstance(lock, Mapping), "spipe.json.lock: must be an object or null.")
        _check_keys(lock, _LOCK_KEYS, "spipe.json.lock")
        for key in ("config_id", "recipe_id"):
            _require(isinstance(lock.get(key), str) and lock[key], f"spipe.json.lock.{key}: required string.")
        _require(lock.get("fit") in ("auto", "in_process"), "spipe.json.lock.fit: must be 'auto' or 'in_process'.")
        _require(isinstance(lock.get("versions"), Mapping), "spipe.json.lock.versions: required object.")

write_manifest(manifest, directory)

Write spipe.json into directory (sorted keys, two-space indent, UTF-8).

Source code in steerability/spipe/format.py

def write_manifest(manifest: Mapping, directory: str | Path) -> None:
    """Write `spipe.json` into `directory` (sorted keys, two-space indent, UTF-8)."""
    path = Path(directory) / MANIFEST_NAME
    with open(path, "w", encoding="utf-8") as handle:
        json.dump(manifest, handle, sort_keys=True, indent=2)
        handle.write("\n")

freeze

Building an SPipe from a live SteeringPipeline (to_spipe orchestration).

The recipe section is the codec encoding of each control's constructor args, in pipeline order. Freezing additionally walks each enabled control's export_state() and frozen_form(), writes the exported artifacts into the bundle's store with provenance and fit digests, and assembles the lock. The freeze walk over every enabled control runs before any recipe encoding, so a recipe value with the same content as an exported artifact reuses the artifact's record.

INLINE_LIMIT_BYTES = 1000000 module-attribute

logger = logging.getLogger(__name__) module-attribute

build_spipe(pipeline, *, freeze=None, model_ref=None)

Build an SPipe from a pipeline.

Parameters:

Name	Type	Description	Default
pipeline	SteeringPipeline	The SteeringPipeline to serialize.	required
freeze	bool | None	Freeze the resolution (default: the pipeline's steered state). False forces a recipe-only spipe from a steered pipeline.	None
model_ref	str | None	Explicit model reference, overriding the pipeline's.	None

Returns:

Type	Description
'SPipe'	The built SPipe (backed by a temporary store directory until saved).

Raises:

Type	Description
SpipeSaveError	If the model reference is unresolvable, a control is unregistered or cannot serialize, freezing is requested on an unsteered pipeline, or a control cannot freeze.

Source code in steerability/spipe/freeze.py

def build_spipe(pipeline: SteeringPipeline, *, freeze: bool | None = None, model_ref: str | None = None) -> "SPipe":
    """Build an `SPipe` from a pipeline.

    Args:
        pipeline: The `SteeringPipeline` to serialize.
        freeze: Freeze the resolution (default: the pipeline's steered state). `False` forces
            a recipe-only spipe from a steered pipeline.
        model_ref: Explicit model reference, overriding the pipeline's.

    Returns:
        The built `SPipe` (backed by a temporary store directory until saved).

    Raises:
        SpipeSaveError: If the model reference is unresolvable, a control is unregistered or
            cannot serialize, freezing is requested on an unsteered pipeline, or a control
            cannot freeze.
    """
    from steerability.algorithms.core.identity import config_descriptor_from_controls, config_digest
    from steerability.algorithms.core.registry import method_key_for
    from steerability.spipe.spipe import SPipe

    if freeze is None:
        freeze = bool(pipeline._is_steered)
    if freeze and not pipeline._is_steered:
        raise SpipeSaveError("freeze=True requires a steered pipeline; call steer() first.")

    ref = _resolve_model_ref(pipeline, model_ref)
    revision = pipeline.hf_model_kwargs.get("revision") if pipeline.hf_model_kwargs else None

    controls = [
        *pipeline.structural_controls, *pipeline.input_controls,
        *pipeline.state_controls, *pipeline.output_controls,
    ]

    temp = tempfile.TemporaryDirectory(prefix="spipe-")
    base_dir = Path(temp.name)
    store = ArtifactStore(base_dir / ARTIFACTS_DIR)

    provenance: dict[str, Any] = {}
    spec = pipeline._resolve_backend_spec(pipeline.backend)
    provenance["backend_spec_hash"] = spec.spec_hash
    model = pipeline.model
    default_model_type = None
    if model is not None:
        from steerability.algorithms.core.internals.fingerprint import artifact_provenance_meta

        meta = artifact_provenance_meta(model, pipeline.tokenizer)
        provenance["model_fingerprint"] = meta.get("model_fingerprint")
        provenance["tokenizer_fingerprint"] = meta.get("chat_template_fingerprint")
        default_model_type = getattr(model.config, "model_type", None)

    ctx = EncodeContext(store=store, provenance=provenance, default_model_type=default_model_type)

    # the freeze walk over every enabled control runs before any recipe encoding: exported
    # artifacts land in the store with their fit provenance first, and a content-equal recipe
    # value of any control reuses the metadata-rich record
    resolved_by_index: dict[int, Any] = {}
    for i, control in enumerate(controls):
        entry_path = f"controls[{i}]"
        if getattr(control, "gate_driven_externally", False):
            raise SpipeSaveError(
                f"{entry_path}: {type(control).__name__} follows an externally driven shared "
                "gate, an in-memory relationship that does not serialize; save the driving "
                "pipeline without the follower, or gate the control directly."
            )
        if freeze and control.enabled:
            try:
                resolved_by_index[i] = _freeze_control(control, ctx, entry_path)
            except NotFreezableError as exc:
                raise SpipeSaveError(
                    f"{entry_path}: {exc} The pipeline can still be saved with freeze=False."
                ) from exc

    entries = []
    for i, control in enumerate(controls):
        entry_path = f"controls[{i}]"
        resolved = resolved_by_index.get(i)
        entry: dict[str, Any] = {
            "method": method_key_for(type(control)),
            "enabled": bool(control.enabled),
            "args": _recipe_args_encoded(control, ctx, f"{entry_path}.args"),
            "resolved": resolved,
        }
        if resolved is not None:
            for item in (resolved if isinstance(resolved, list) else [resolved]):
                if item.get("origin") is not None:
                    item["origin"]["args"] = entry["args"]
        size = encoded_size(entry)
        if size > INLINE_LIMIT_BYTES:
            raise SpipeSaveError(
                f"{entry_path}: the entry's inline JSON is {size} bytes, over the "
                f"{INLINE_LIMIT_BYTES}-byte limit; convert large dataset fields to a $data "
                "reference (DataRef), or save the fitted pipeline frozen and share that."
            )
        entries.append(entry)

    descriptor = config_descriptor_from_controls(controls)
    config_id = config_digest(descriptor)
    recipe_id = config_digest({"model": ref, "controls": descriptor["controls"]})

    lock = None
    if freeze:
        lock = {
            "config_id": config_id,
            "recipe_id": recipe_id,
            "model_fingerprint": provenance.get("model_fingerprint"),
            "tokenizer_fingerprint": provenance.get("tokenizer_fingerprint"),
            "torch_dtype": str(model.dtype).removeprefix("torch.") if model is not None else None,
            "steer_backend_spec_hash": spec.spec_hash,
            "fit": pipeline.fit,
            "seed": None,
            "versions": _package_versions(),
        }

    manifest = {
        "format": FORMAT_VERSION,
        "created_at": datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"),
        "toolkit_version": _toolkit_version(),
        "code_dependent": bool(ctx.code_refs),
        "model": {"ref": ref, "revision": revision},
        "controls": entries,
        "lock": lock,
    }

    return SPipe(manifest, store=store, base_dir=base_dir, allow_code=True, _temp=temp)

spipe

SPipe: a portable serialization of a SteeringPipeline.

One format holds both the recipe and the frozen resolution. A recipe-only spipe carries the model reference and the controls as constructed; loading it and calling steer() re-runs fits. A frozen spipe additionally pins what the resolution produced (fingerprints, resolved bindings, per-fit digests) in a lock section and stores the products content-addressed. The loaded controls therefore steer cheaply and model-free. The frozen form is itself a valid recipe, since every resolved entry is constructor-valid for its method and loading takes the ordinary construction and steer() path.

logger = logging.getLogger(__name__) module-attribute

SPipe

A serialized steering pipeline: recipe, optional lock, and artifact store.

Instances come from SteeringPipeline.to_spipe() (backed by a temporary store until saved) or SPipe.load() (backed by the loaded bundle). The manifest is data; controls are instantiated only by pipeline().

Source code in steerability/spipe/spipe.py

class SPipe:
    """A serialized steering pipeline: recipe, optional lock, and artifact store.

    Instances come from `SteeringPipeline.to_spipe()` (backed by a temporary store until
    saved) or `SPipe.load()` (backed by the loaded bundle). The manifest is data; controls
    are instantiated only by `pipeline()`.
    """

    def __init__(
        self,
        manifest: dict,
        *,
        store: ArtifactStore | None,
        base_dir: Path | None,
        allow_code: bool,
        allow_stale: bool = False,
        _temp: Any = None,
    ):
        validate_manifest(manifest)
        self._manifest = manifest
        self._manifest_records = _manifest_records(manifest)
        self._store = store
        self._base_dir = base_dir
        self._allow_code = allow_code
        self._allow_stale = allow_stale
        self._temp = _temp
        self._staleness_checked = False

    # inspection

    @property
    def manifest(self) -> dict:
        """A deep copy of the manifest."""
        return copy.deepcopy(self._manifest)

    @property
    def recipe_id(self) -> str:
        """The recipe identity digest (model reference plus controls), 12 hex characters."""
        lock = self._manifest.get("lock")
        if lock is not None:
            return lock["recipe_id"]
        from steerability.algorithms.core.identity import config_digest

        return config_digest({
            "model": self._manifest["model"]["ref"],
            "controls": self._descriptor()["controls"],
        })

    @property
    def config_id(self) -> str:
        """The configuration identity digest over the recipe entries, 12 hex characters."""
        lock = self._manifest.get("lock")
        if lock is not None:
            return lock["config_id"]
        from steerability.algorithms.core.identity import config_digest

        return config_digest(self._descriptor())

    def _descriptor(self) -> dict:
        """The configuration descriptor recomputed from freshly instantiated recipe controls."""
        from steerability.algorithms.core.identity import config_descriptor_from_controls

        controls = [self._instantiate(entry["method"], entry["args"], self._decode_ctx(lenient=True))
                    for entry in self._manifest["controls"]]
        for control, entry in zip(controls, self._manifest["controls"]):
            control.enabled = entry["enabled"]
        return config_descriptor_from_controls(controls)

    @property
    def code_dependent(self) -> bool:
        """Whether loading this spipe needs matching code and `allow_code=True`."""
        return bool(self._manifest["code_dependent"])

    @property
    def is_frozen(self) -> bool:
        """True iff every enabled entry with steer-time products carries a resolution.

        Entries with `resolved: null` have nothing to pin (their recipe is their frozen
        form), which makes a manifest with a lock section frozen.
        """
        return self._manifest.get("lock") is not None

    def describe(self) -> str:
        """A human-readable table of entries, frozen state, and artifact sizes."""
        lines = [
            f"spipe {self._manifest['format']}  model={self._manifest['model']['ref']}",
            f"  recipe_id={self.recipe_id}  config_id={self.config_id}  "
            f"frozen={self.is_frozen}  code_dependent={self.code_dependent}",
        ]
        for i, entry in enumerate(self._manifest["controls"]):
            items = _resolved_items(entry)
            frozen = "recipe"
            if items:
                methods = ", ".join(item["method"] for item in items)
                frozen = f"frozen -> {methods}"
            elif self.is_frozen and entry["enabled"]:
                frozen = "frozen (recipe is the frozen form)"
            enabled = "" if entry["enabled"] else "  [disabled]"
            lines.append(f"  [{i}] {entry['method']}{enabled}  ({frozen})")
            for item in items:
                for name, record in (item.get("artifacts") or {}).items():
                    size = ""
                    if self._store is not None and self._store.has(record["id"]):
                        size = f"  {self._store.size_of(record['id'])} bytes"
                    lines.append(f"        {name}: {record['type']} {record['id'][:19]}…{size}")
        return "\n".join(lines)

    # freeze state

    def thaw(self) -> "SPipe":
        """A recipe-only copy: every `resolved` section and the lock are dropped."""
        manifest = self.manifest
        for entry in manifest["controls"]:
            entry["resolved"] = None
        manifest["lock"] = None
        return SPipe(
            manifest, store=self._store, base_dir=self._base_dir,
            allow_code=self._allow_code, allow_stale=self._allow_stale, _temp=self._temp,
        )

    # verification

    def _referenced_artifact_ids(self) -> set[str]:
        found: set[str] = set()
        _collect_artifact_ids(self._manifest["controls"], found)
        return found

    def verify(self) -> SpipeReport:
        """The model-free report: format, integrity, staleness, versions, code dependence.

        Never loads a model or instantiates a backend. Referenced artifacts that are not
        present are reported in one warning (a thin bundle); the present ones are verified
        against their content ids.

        Returns:
            The report.
        """
        errors: list[str] = []
        report_warnings: list[str] = []

        try:
            validate_manifest(self._manifest)
        except SpipeFormatError as exc:
            errors.append(str(exc))

        referenced = sorted(self._referenced_artifact_ids())
        missing = [
            artifact_id for artifact_id in referenced
            if self._store is None or not self._store.has(artifact_id)
        ]
        if missing:
            report_warnings.append(
                f"{len(missing)} of {len(referenced)} referenced artifact(s) are not present (thin "
                f"bundle); pass artifact_store= at load to resolve them: {', '.join(missing)}."
            )
        for artifact_id in referenced:
            if artifact_id not in missing:
                try:
                    self._store.verify(artifact_id)
                except SpipeIntegrityError as exc:
                    errors.append(str(exc))

        try:
            self._check_staleness(raise_on_stale=True, force=True)
        except SpipeStaleError as exc:
            errors.append(str(exc))
        except Exception as exc:
            report_warnings.append(f"staleness check could not run: {exc}")

        from steerability.spipe.freeze import _toolkit_version

        saved = self._manifest.get("toolkit_version", "unknown")
        current = _toolkit_version()
        if saved.split(".")[0] != current.split(".")[0]:
            report_warnings.append(
                f"spipe was written by steerability {saved}; this is {current} (major mismatch)."
            )
        if self.code_dependent:
            report_warnings.append(
                "the manifest references code ($ref); loading a pipeline from it requires "
                "allow_code=True and the referenced modules on the import path."
            )

        return SpipeReport(ok=not errors, errors=errors, warnings=report_warnings)

    def _decode_ctx(self, *, lenient: bool, verify: str = "off", data_mode: str = "keep") -> DecodeContext:
        return DecodeContext(
            store=self._store,
            allow_code=self._allow_code,
            code_mode="sentinel" if lenient else "strict",
            verify=verify,
            data_mode=data_mode,
            manifest_records=self._manifest_records,
        )

    @staticmethod
    def _instantiate(method_key: str, encoded_args: Mapping, ctx: DecodeContext):
        from steerability.algorithms.core.registry import RegistryError, resolve_method_key

        try:
            method = resolve_method_key(method_key)
        except RegistryError as exc:
            raise SpipeFormatError(str(exc)) from exc
        kwargs = {name: decode(value, ctx, f"args.{name}") for name, value in encoded_args.items()}
        return method.control_cls(**kwargs)

    def _check_staleness(self, *, raise_on_stale: bool, force: bool = False) -> None:
        """Recompute each frozen entry's fit digest from the current recipe args.

        Skipped when artifacts are unavailable (thin bundle without a store); `pipeline()`
        re-invokes it once artifacts resolve.

        Raises:
            SpipeStaleError: If a recorded fit digest does not match the recomputed one, or a
                recipe entry can no longer be reconstructed for the check.
        """
        if self._staleness_checked and not force:
            return
        for i, entry in enumerate(self._manifest["controls"]):
            recorded: dict[str, str] = {}
            for item in _resolved_items(entry):
                for name, record in (item.get("artifacts") or {}).items():
                    if record.get("fit_digest"):
                        recorded[name] = record["fit_digest"]
            if not recorded:
                continue
            try:
                control = self._instantiate(entry["method"], entry["args"], self._decode_ctx(lenient=True))
                fit_identity = control.fit_identity()
            except Exception as exc:
                raise SpipeStaleError(
                    f"controls[{i}] ({entry['method']}): the recipe args no longer reconstruct "
                    f"for the staleness check ({exc}); thaw() and re-steer(), or pass "
                    "allow_stale=True."
                ) from exc
            current = digest_of(fit_identity) if fit_identity is not None else None
            for name, digest in recorded.items():
                if digest != current:
                    message = (
                        f"controls[{i}] ({entry['method']}): frozen artifact {name!r} was "
                        f"produced from fit digest {digest} but the recipe now digests to "
                        f"{current}; the fit-relevant recipe fields were edited after "
                        "freezing. thaw() and re-steer(), or pass allow_stale=True."
                    )
                    if raise_on_stale:
                        raise SpipeStaleError(message)
                    warnings.warn(message, UserWarning)
        self._staleness_checked = True

    # load / save

    @classmethod
    def load(
        cls,
        path: str | Path,
        *,
        allow_code: bool = False,
        allow_stale: bool = False,
        artifact_store: str | Path | Callable[[str], Path] | None = None,
    ) -> "SPipe":
        """Load a `.spipe` file or directory.

        Args:
            path: A `.spipe` zip file or a spipe directory.
            allow_code: Permit `$ref` imports, non-`steerability.` `$dc` imports, and
                pickle-backed memory payloads during decoding.
            allow_stale: Skip the staleness check.
            artifact_store: External artifact source for thin bundles: a store directory, or
                a callable mapping an artifact id to the directory holding it.

        Returns:
            The loaded `SPipe`.

        Raises:
            SpipeFormatError: On format-version, schema, or archive violations.
            SpipeIntegrityError: If a present artifact fails content verification.
            SpipeStaleError: If a frozen entry is stale and `allow_stale` is False.
        """
        path = Path(path)
        temp = None
        if path.is_dir():
            base_dir = path
        else:
            temp = tempfile.TemporaryDirectory(prefix="spipe-load-")
            unpack_zip(path, temp.name)
            base_dir = Path(temp.name)

        manifest = read_manifest(base_dir)

        resolver: Callable[[str], Path] | None = None
        if callable(artifact_store):
            resolver = artifact_store
        elif artifact_store is not None:
            external = Path(artifact_store)
            resolver = lambda artifact_id: external / artifact_id.replace(":", "-", 1)  # noqa: E731
        store = ArtifactStore(base_dir / ARTIFACTS_DIR, resolver=resolver)

        spipe = cls(
            manifest, store=store, base_dir=base_dir,
            allow_code=allow_code, allow_stale=allow_stale, _temp=temp,
        )

        referenced = spipe._referenced_artifact_ids()
        available = [artifact_id for artifact_id in referenced if store.has(artifact_id)]
        for artifact_id in sorted(available):
            store.verify(artifact_id)
        if len(available) < len(referenced):
            logger.info(
                "Thin spipe: %d of %d referenced artifacts unavailable; integrity and "
                "staleness checks defer to pipeline construction.",
                len(referenced) - len(available), len(referenced),
            )
        elif not allow_stale:
            spipe._check_staleness(raise_on_stale=True)

        return spipe

    def save(self, path: str | Path, *, artifacts: str = "fat") -> Path:
        """Write the spipe to `path`.

        A path ending in `.spipe` produces a zip file; any other path produces (or replaces)
        a directory. `artifacts="fat"` (default) embeds every referenced artifact;
        `artifacts="thin"` writes the manifest only, leaving artifact ids resolvable at load
        via `artifact_store=`. A recipe-only spipe with no artifact references writes no
        `artifacts/` directory either way.

        Saving onto the bundle's own backing directory rewrites the manifest in place. A fat
        save there also embeds any referenced artifact the store resolves externally, and a
        thin save there raises `SpipeSaveError` when the directory embeds artifacts, since a
        thin export would have to delete them.

        Args:
            path: Destination file or directory.
            artifacts: `"fat"` or `"thin"`.

        Returns:
            The written path.

        Raises:
            SpipeSaveError: If a directory target exists and is neither empty nor a spipe
                directory, a referenced artifact is unavailable for a fat export, or a thin
                export targets the bundle's own directory while it embeds artifacts.
        """
        if artifacts not in ("fat", "thin"):
            raise SpipeSaveError(f"artifacts must be 'fat' or 'thin'; got {artifacts!r}.")
        path = Path(path)
        referenced = sorted(self._referenced_artifact_ids())
        if artifacts == "fat" and referenced:
            if self._store is None:
                raise SpipeSaveError("This spipe has no artifact store; only artifacts='thin' is possible.")
            for artifact_id in referenced:
                self._store.verify(artifact_id)

        if path.suffix == ".spipe":
            with tempfile.TemporaryDirectory(prefix="spipe-save-") as staging:
                write_manifest(self._manifest, staging)
                if artifacts == "fat" and referenced:
                    self._store.copy_into(Path(staging) / ARTIFACTS_DIR, referenced)
                path.parent.mkdir(parents=True, exist_ok=True)
                pack_zip(staging, path)
            return path

        if self._base_dir is not None and path.exists() \
                and path.resolve() == Path(self._base_dir).resolve():
            # saving onto the backing directory rewrites the manifest in place; a fat save also
            # embeds artifacts the store resolves externally, and a thin save is refused when the
            # directory embeds artifacts since it would have to delete them
            if artifacts == "thin" and (path / ARTIFACTS_DIR).is_dir():
                raise SpipeSaveError(
                    f"A thin export onto the bundle's own directory {path} would delete its "
                    "embedded artifacts; save the thin export to a new path."
                )
            write_manifest(self._manifest, path)
            if artifacts == "fat" and referenced:
                self._store.copy_into(path / ARTIFACTS_DIR, referenced)
            return path

        if path.exists():
            if not path.is_dir():
                raise SpipeSaveError(f"{path} exists and is not a directory.")
            occupied = any(path.iterdir())
            if occupied and not (path / MANIFEST_NAME).exists():
                raise SpipeSaveError(
                    f"{path} exists and is not a spipe directory; refusing to replace it."
                )
            for member in (path / MANIFEST_NAME, path / ARTIFACTS_DIR):
                if member.is_dir():
                    shutil.rmtree(member)
                elif member.exists():
                    member.unlink()
        path.mkdir(parents=True, exist_ok=True)
        write_manifest(self._manifest, path)
        if artifacts == "fat" and referenced:
            self._store.copy_into(path / ARTIFACTS_DIR, referenced)
        return path

    # pipeline construction

    def pipeline(
        self,
        *,
        backend: BackendSpec | str | None = None,
        prefer: str = "frozen",
        verify: str = "strict",
        **pipeline_kwargs,
    ) -> SteeringPipeline:
        """Instantiate a `SteeringPipeline` from this spipe.

        Frozen entries instantiate from their resolution by default; `prefer="recipe"`
        instantiates every entry from its recipe args (forcing re-fits at `steer()`). The
        spipe supplies the model reference and the controls only; backend, device, dtype, and
        `hf_model_kwargs` are the caller's. The caller runs `pipeline.check()` /
        `pipeline.steer()` as normal.

        Args:
            backend: Forwarded to the `SteeringPipeline` constructor.
            prefer: `"frozen"` (default) or `"recipe"`.
            verify: Verification policy for frozen steering artifacts (`"strict"`, `"warn"`,
                or `"off"`), enforced where binding happens at `steer()`.

        Returns:
            The constructed (unsteered) `SteeringPipeline`.

        Raises:
            SpipeFormatError: If a method key resolves to no registered method.
            SpipeCodeRefError: If decoding requires code and the spipe was loaded without
                `allow_code=True`.
            SpipeStaleError: If a deferred staleness check fails (thin bundles).
        """
        if prefer not in ("frozen", "recipe"):
            raise ValueError(f"prefer must be 'frozen' or 'recipe'; got {prefer!r}.")
        if verify not in ("strict", "warn", "off"):
            raise ValueError(f"verify must be 'strict', 'warn', or 'off'; got {verify!r}.")

        if prefer == "frozen" and not self._allow_stale:
            self._check_staleness(raise_on_stale=True)

        ctx = DecodeContext(
            store=self._store,
            allow_code=self._allow_code,
            code_mode="strict",
            verify=verify,
            data_mode="load",
            manifest_records=self._manifest_records,
        )

        controls = []
        for entry in self._manifest["controls"]:
            items = _resolved_items(entry) if prefer == "frozen" else []
            if items:
                for item in items:
                    args = dict(item["args"])
                    if verify != "strict" and item["method"] == "output_control/routed_decoding":
                        args["allow_model_mismatch"] = True
                    if verify != "strict" and item["method"] == "structural_control/load_lora":
                        args["allow_base_mismatch"] = True
                    control = self._instantiate(item["method"], args, ctx)
                    control.enabled = entry["enabled"]
                    controls.append(control)
            else:
                control = self._instantiate(entry["method"], entry["args"], ctx)
                control.enabled = entry["enabled"]
                controls.append(control)

        # controls may hold paths into this spipe's extraction directory; retaining the spipe
        # on each control keeps that directory alive for the pipeline's lifetime
        for control in controls:
            control._spipe_retainer = self

        from steerability.algorithms.core.steering_pipeline import SteeringPipeline

        return SteeringPipeline(
            model_name_or_path=self._manifest["model"]["ref"],
            controls=controls,
            backend=backend,
            **pipeline_kwargs,
        )

code_dependent property

Whether loading this spipe needs matching code and allow_code=True.

config_id property

The configuration identity digest over the recipe entries, 12 hex characters.

is_frozen property

True iff every enabled entry with steer-time products carries a resolution.

Entries with resolved: null have nothing to pin (their recipe is their frozen form), which makes a manifest with a lock section frozen.

manifest property

A deep copy of the manifest.

recipe_id property

The recipe identity digest (model reference plus controls), 12 hex characters.

describe()

A human-readable table of entries, frozen state, and artifact sizes.

Source code in steerability/spipe/spipe.py

def describe(self) -> str:
    """A human-readable table of entries, frozen state, and artifact sizes."""
    lines = [
        f"spipe {self._manifest['format']}  model={self._manifest['model']['ref']}",
        f"  recipe_id={self.recipe_id}  config_id={self.config_id}  "
        f"frozen={self.is_frozen}  code_dependent={self.code_dependent}",
    ]
    for i, entry in enumerate(self._manifest["controls"]):
        items = _resolved_items(entry)
        frozen = "recipe"
        if items:
            methods = ", ".join(item["method"] for item in items)
            frozen = f"frozen -> {methods}"
        elif self.is_frozen and entry["enabled"]:
            frozen = "frozen (recipe is the frozen form)"
        enabled = "" if entry["enabled"] else "  [disabled]"
        lines.append(f"  [{i}] {entry['method']}{enabled}  ({frozen})")
        for item in items:
            for name, record in (item.get("artifacts") or {}).items():
                size = ""
                if self._store is not None and self._store.has(record["id"]):
                    size = f"  {self._store.size_of(record['id'])} bytes"
                lines.append(f"        {name}: {record['type']} {record['id'][:19]}…{size}")
    return "\n".join(lines)

load(path, *, allow_code=False, allow_stale=False, artifact_store=None) classmethod

Load a .spipe file or directory.

Parameters:

Name	Type	Description	Default
path	str | Path	A .spipe zip file or a spipe directory.	required
allow_code	bool	Permit $ref imports, non-steerability. $dc imports, and pickle-backed memory payloads during decoding.	False
allow_stale	bool	Skip the staleness check.	False
artifact_store	str | Path | Callable[[str], Path] | None	External artifact source for thin bundles: a store directory, or a callable mapping an artifact id to the directory holding it.	None

Returns:

Type	Description
'SPipe'	The loaded SPipe.

Raises:

Type	Description
SpipeFormatError	On format-version, schema, or archive violations.
SpipeIntegrityError	If a present artifact fails content verification.
SpipeStaleError	If a frozen entry is stale and allow_stale is False.

Source code in steerability/spipe/spipe.py

@classmethod
def load(
    cls,
    path: str | Path,
    *,
    allow_code: bool = False,
    allow_stale: bool = False,
    artifact_store: str | Path | Callable[[str], Path] | None = None,
) -> "SPipe":
    """Load a `.spipe` file or directory.

    Args:
        path: A `.spipe` zip file or a spipe directory.
        allow_code: Permit `$ref` imports, non-`steerability.` `$dc` imports, and
            pickle-backed memory payloads during decoding.
        allow_stale: Skip the staleness check.
        artifact_store: External artifact source for thin bundles: a store directory, or
            a callable mapping an artifact id to the directory holding it.

    Returns:
        The loaded `SPipe`.

    Raises:
        SpipeFormatError: On format-version, schema, or archive violations.
        SpipeIntegrityError: If a present artifact fails content verification.
        SpipeStaleError: If a frozen entry is stale and `allow_stale` is False.
    """
    path = Path(path)
    temp = None
    if path.is_dir():
        base_dir = path
    else:
        temp = tempfile.TemporaryDirectory(prefix="spipe-load-")
        unpack_zip(path, temp.name)
        base_dir = Path(temp.name)

    manifest = read_manifest(base_dir)

    resolver: Callable[[str], Path] | None = None
    if callable(artifact_store):
        resolver = artifact_store
    elif artifact_store is not None:
        external = Path(artifact_store)
        resolver = lambda artifact_id: external / artifact_id.replace(":", "-", 1)  # noqa: E731
    store = ArtifactStore(base_dir / ARTIFACTS_DIR, resolver=resolver)

    spipe = cls(
        manifest, store=store, base_dir=base_dir,
        allow_code=allow_code, allow_stale=allow_stale, _temp=temp,
    )

    referenced = spipe._referenced_artifact_ids()
    available = [artifact_id for artifact_id in referenced if store.has(artifact_id)]
    for artifact_id in sorted(available):
        store.verify(artifact_id)
    if len(available) < len(referenced):
        logger.info(
            "Thin spipe: %d of %d referenced artifacts unavailable; integrity and "
            "staleness checks defer to pipeline construction.",
            len(referenced) - len(available), len(referenced),
        )
    elif not allow_stale:
        spipe._check_staleness(raise_on_stale=True)

    return spipe

pipeline(*, backend=None, prefer='frozen', verify='strict', **pipeline_kwargs)

Instantiate a SteeringPipeline from this spipe.

Frozen entries instantiate from their resolution by default; prefer="recipe" instantiates every entry from its recipe args (forcing re-fits at steer()). The spipe supplies the model reference and the controls only; backend, device, dtype, and hf_model_kwargs are the caller's. The caller runs pipeline.check() / pipeline.steer() as normal.

Parameters:

Name	Type	Description	Default
backend	BackendSpec | str | None	Forwarded to the SteeringPipeline constructor.	None
prefer	str	"frozen" (default) or "recipe".	'frozen'
verify	str	Verification policy for frozen steering artifacts ("strict", "warn", or "off"), enforced where binding happens at steer().	'strict'

Returns:

Type	Description
SteeringPipeline	The constructed (unsteered) SteeringPipeline.

Raises:

Type	Description
SpipeFormatError	If a method key resolves to no registered method.
SpipeCodeRefError	If decoding requires code and the spipe was loaded without allow_code=True.
SpipeStaleError	If a deferred staleness check fails (thin bundles).

Source code in steerability/spipe/spipe.py

def pipeline(
    self,
    *,
    backend: BackendSpec | str | None = None,
    prefer: str = "frozen",
    verify: str = "strict",
    **pipeline_kwargs,
) -> SteeringPipeline:
    """Instantiate a `SteeringPipeline` from this spipe.

    Frozen entries instantiate from their resolution by default; `prefer="recipe"`
    instantiates every entry from its recipe args (forcing re-fits at `steer()`). The
    spipe supplies the model reference and the controls only; backend, device, dtype, and
    `hf_model_kwargs` are the caller's. The caller runs `pipeline.check()` /
    `pipeline.steer()` as normal.

    Args:
        backend: Forwarded to the `SteeringPipeline` constructor.
        prefer: `"frozen"` (default) or `"recipe"`.
        verify: Verification policy for frozen steering artifacts (`"strict"`, `"warn"`,
            or `"off"`), enforced where binding happens at `steer()`.

    Returns:
        The constructed (unsteered) `SteeringPipeline`.

    Raises:
        SpipeFormatError: If a method key resolves to no registered method.
        SpipeCodeRefError: If decoding requires code and the spipe was loaded without
            `allow_code=True`.
        SpipeStaleError: If a deferred staleness check fails (thin bundles).
    """
    if prefer not in ("frozen", "recipe"):
        raise ValueError(f"prefer must be 'frozen' or 'recipe'; got {prefer!r}.")
    if verify not in ("strict", "warn", "off"):
        raise ValueError(f"verify must be 'strict', 'warn', or 'off'; got {verify!r}.")

    if prefer == "frozen" and not self._allow_stale:
        self._check_staleness(raise_on_stale=True)

    ctx = DecodeContext(
        store=self._store,
        allow_code=self._allow_code,
        code_mode="strict",
        verify=verify,
        data_mode="load",
        manifest_records=self._manifest_records,
    )

    controls = []
    for entry in self._manifest["controls"]:
        items = _resolved_items(entry) if prefer == "frozen" else []
        if items:
            for item in items:
                args = dict(item["args"])
                if verify != "strict" and item["method"] == "output_control/routed_decoding":
                    args["allow_model_mismatch"] = True
                if verify != "strict" and item["method"] == "structural_control/load_lora":
                    args["allow_base_mismatch"] = True
                control = self._instantiate(item["method"], args, ctx)
                control.enabled = entry["enabled"]
                controls.append(control)
        else:
            control = self._instantiate(entry["method"], entry["args"], ctx)
            control.enabled = entry["enabled"]
            controls.append(control)

    # controls may hold paths into this spipe's extraction directory; retaining the spipe
    # on each control keeps that directory alive for the pipeline's lifetime
    for control in controls:
        control._spipe_retainer = self

    from steerability.algorithms.core.steering_pipeline import SteeringPipeline

    return SteeringPipeline(
        model_name_or_path=self._manifest["model"]["ref"],
        controls=controls,
        backend=backend,
        **pipeline_kwargs,
    )

save(path, *, artifacts='fat')

Write the spipe to path.

A path ending in .spipe produces a zip file; any other path produces (or replaces) a directory. artifacts="fat" (default) embeds every referenced artifact; artifacts="thin" writes the manifest only, leaving artifact ids resolvable at load via artifact_store=. A recipe-only spipe with no artifact references writes no artifacts/ directory either way.

Saving onto the bundle's own backing directory rewrites the manifest in place. A fat save there also embeds any referenced artifact the store resolves externally, and a thin save there raises SpipeSaveError when the directory embeds artifacts, since a thin export would have to delete them.

Parameters:

Name	Type	Description	Default
path	str | Path	Destination file or directory.	required
artifacts	str	"fat" or "thin".	'fat'

Returns:

Type	Description
Path	The written path.

Raises:

Type	Description
SpipeSaveError	If a directory target exists and is neither empty nor a spipe directory, a referenced artifact is unavailable for a fat export, or a thin export targets the bundle's own directory while it embeds artifacts.

Source code in steerability/spipe/spipe.py

def save(self, path: str | Path, *, artifacts: str = "fat") -> Path:
    """Write the spipe to `path`.

    A path ending in `.spipe` produces a zip file; any other path produces (or replaces)
    a directory. `artifacts="fat"` (default) embeds every referenced artifact;
    `artifacts="thin"` writes the manifest only, leaving artifact ids resolvable at load
    via `artifact_store=`. A recipe-only spipe with no artifact references writes no
    `artifacts/` directory either way.

    Saving onto the bundle's own backing directory rewrites the manifest in place. A fat
    save there also embeds any referenced artifact the store resolves externally, and a
    thin save there raises `SpipeSaveError` when the directory embeds artifacts, since a
    thin export would have to delete them.

    Args:
        path: Destination file or directory.
        artifacts: `"fat"` or `"thin"`.

    Returns:
        The written path.

    Raises:
        SpipeSaveError: If a directory target exists and is neither empty nor a spipe
            directory, a referenced artifact is unavailable for a fat export, or a thin
            export targets the bundle's own directory while it embeds artifacts.
    """
    if artifacts not in ("fat", "thin"):
        raise SpipeSaveError(f"artifacts must be 'fat' or 'thin'; got {artifacts!r}.")
    path = Path(path)
    referenced = sorted(self._referenced_artifact_ids())
    if artifacts == "fat" and referenced:
        if self._store is None:
            raise SpipeSaveError("This spipe has no artifact store; only artifacts='thin' is possible.")
        for artifact_id in referenced:
            self._store.verify(artifact_id)

    if path.suffix == ".spipe":
        with tempfile.TemporaryDirectory(prefix="spipe-save-") as staging:
            write_manifest(self._manifest, staging)
            if artifacts == "fat" and referenced:
                self._store.copy_into(Path(staging) / ARTIFACTS_DIR, referenced)
            path.parent.mkdir(parents=True, exist_ok=True)
            pack_zip(staging, path)
        return path

    if self._base_dir is not None and path.exists() \
            and path.resolve() == Path(self._base_dir).resolve():
        # saving onto the backing directory rewrites the manifest in place; a fat save also
        # embeds artifacts the store resolves externally, and a thin save is refused when the
        # directory embeds artifacts since it would have to delete them
        if artifacts == "thin" and (path / ARTIFACTS_DIR).is_dir():
            raise SpipeSaveError(
                f"A thin export onto the bundle's own directory {path} would delete its "
                "embedded artifacts; save the thin export to a new path."
            )
        write_manifest(self._manifest, path)
        if artifacts == "fat" and referenced:
            self._store.copy_into(path / ARTIFACTS_DIR, referenced)
        return path

    if path.exists():
        if not path.is_dir():
            raise SpipeSaveError(f"{path} exists and is not a directory.")
        occupied = any(path.iterdir())
        if occupied and not (path / MANIFEST_NAME).exists():
            raise SpipeSaveError(
                f"{path} exists and is not a spipe directory; refusing to replace it."
            )
        for member in (path / MANIFEST_NAME, path / ARTIFACTS_DIR):
            if member.is_dir():
                shutil.rmtree(member)
            elif member.exists():
                member.unlink()
    path.mkdir(parents=True, exist_ok=True)
    write_manifest(self._manifest, path)
    if artifacts == "fat" and referenced:
        self._store.copy_into(path / ARTIFACTS_DIR, referenced)
    return path

thaw()

A recipe-only copy: every resolved section and the lock are dropped.

Source code in steerability/spipe/spipe.py

def thaw(self) -> "SPipe":
    """A recipe-only copy: every `resolved` section and the lock are dropped."""
    manifest = self.manifest
    for entry in manifest["controls"]:
        entry["resolved"] = None
    manifest["lock"] = None
    return SPipe(
        manifest, store=self._store, base_dir=self._base_dir,
        allow_code=self._allow_code, allow_stale=self._allow_stale, _temp=self._temp,
    )

verify()

The model-free report: format, integrity, staleness, versions, code dependence.

Never loads a model or instantiates a backend. Referenced artifacts that are not present are reported in one warning (a thin bundle); the present ones are verified against their content ids.

Returns:

Type	Description
SpipeReport	The report.

Source code in steerability/spipe/spipe.py

def verify(self) -> SpipeReport:
    """The model-free report: format, integrity, staleness, versions, code dependence.

    Never loads a model or instantiates a backend. Referenced artifacts that are not
    present are reported in one warning (a thin bundle); the present ones are verified
    against their content ids.

    Returns:
        The report.
    """
    errors: list[str] = []
    report_warnings: list[str] = []

    try:
        validate_manifest(self._manifest)
    except SpipeFormatError as exc:
        errors.append(str(exc))

    referenced = sorted(self._referenced_artifact_ids())
    missing = [
        artifact_id for artifact_id in referenced
        if self._store is None or not self._store.has(artifact_id)
    ]
    if missing:
        report_warnings.append(
            f"{len(missing)} of {len(referenced)} referenced artifact(s) are not present (thin "
            f"bundle); pass artifact_store= at load to resolve them: {', '.join(missing)}."
        )
    for artifact_id in referenced:
        if artifact_id not in missing:
            try:
                self._store.verify(artifact_id)
            except SpipeIntegrityError as exc:
                errors.append(str(exc))

    try:
        self._check_staleness(raise_on_stale=True, force=True)
    except SpipeStaleError as exc:
        errors.append(str(exc))
    except Exception as exc:
        report_warnings.append(f"staleness check could not run: {exc}")

    from steerability.spipe.freeze import _toolkit_version

    saved = self._manifest.get("toolkit_version", "unknown")
    current = _toolkit_version()
    if saved.split(".")[0] != current.split(".")[0]:
        report_warnings.append(
            f"spipe was written by steerability {saved}; this is {current} (major mismatch)."
        )
    if self.code_dependent:
        report_warnings.append(
            "the manifest references code ($ref); loading a pipeline from it requires "
            "allow_code=True and the referenced modules on the import path."
        )

    return SpipeReport(ok=not errors, errors=errors, warnings=report_warnings)

SpipeReport dataclass

The result of a model-free SPipe.verify().

Attributes:

Name	Type	Description
ok	bool	True when no errors were found.
errors	list[str]	Findings that make the bundle unusable as-is.
warnings	list[str]	Findings worth knowing that do not block loading.

Source code in steerability/spipe/spipe.py

@dataclass
class SpipeReport:
    """The result of a model-free `SPipe.verify()`.

    Attributes:
        ok: True when no errors were found.
        errors: Findings that make the bundle unusable as-is.
        warnings: Findings worth knowing that do not block loading.
    """

    ok: bool
    errors: list[str] = field(default_factory=list)
    warnings: list[str] = field(default_factory=list)

    def render(self) -> str:
        """A human-readable summary."""
        lines = [f"spipe verify: {'ok' if self.ok else 'FAILED'}"]
        lines.extend(f"  error: {message}" for message in self.errors)
        lines.extend(f"  warning: {message}" for message in self.warnings)
        return "\n".join(lines)

    def __str__(self) -> str:
        return self.render()

errors = field(default_factory=list) class-attribute instance-attribute

ok instance-attribute

warnings = field(default_factory=list) class-attribute instance-attribute

render()

A human-readable summary.

Source code in steerability/spipe/spipe.py

def render(self) -> str:
    """A human-readable summary."""
    lines = [f"spipe verify: {'ok' if self.ok else 'FAILED'}"]
    lines.extend(f"  error: {message}" for message in self.errors)
    lines.extend(f"  warning: {message}" for message in self.warnings)
    return "\n".join(lines)

store

Content-addressed artifact store backing frozen .spipe bundles.

The store is a directory of artifact directories, each named by its content id. Two encodings exist: "tensors" (a single artifact.safetensors, id and bytes produced by artifact_id_for from state_control/common/lowering.py, byte-compatible with the vLLM-Hook plugin registry) and "tree" (a directory copied verbatim under payload/, id a SHA-256 over the sorted relative paths and per-file digests). Every artifact directory carries an artifact.json sidecar duplicating its manifest record plus type-specific reconstruction metadata, which keeps a detached artifact directory self-describing. Writes are idempotent and reads verify the content hash.

PAYLOAD_DIR = 'payload' module-attribute

SIDECAR_FILE = 'artifact.json' module-attribute

TENSOR_FILE = 'artifact.safetensors' module-attribute

logger = logging.getLogger(__name__) module-attribute

ArtifactRecord dataclass

One artifact's manifest record.

Attributes:

Name	Type	Description
id	str	Content-addressed artifact id (sha256:<hex>).
encoding	str	"tensors" or "tree".
type	str	Name of the producing class (e.g. "SteeringVector", "Probe").
artifact_class	str	"direction", "calibrated", or "opaque".
source	str | None	Fit-source class name, or None for recipe-supplied artifacts.
fit_digest	str | None	12-hex-character digest of the producing fit identity, or None.
provenance	dict	Producing-side fingerprints (backend_spec_hash, model_fingerprint, tokenizer_fingerprint), each possibly None.
type_meta	dict	Type-specific reconstruction metadata (sidecar only, excluded from the manifest record).

Source code in steerability/spipe/store.py

@dataclass
class ArtifactRecord:
    """One artifact's manifest record.

    Attributes:
        id: Content-addressed artifact id (`sha256:<hex>`).
        encoding: `"tensors"` or `"tree"`.
        type: Name of the producing class (e.g. `"SteeringVector"`, `"Probe"`).
        artifact_class: `"direction"`, `"calibrated"`, or `"opaque"`.
        source: Fit-source class name, or None for recipe-supplied artifacts.
        fit_digest: 12-hex-character digest of the producing fit identity, or None.
        provenance: Producing-side fingerprints (`backend_spec_hash`, `model_fingerprint`,
            `tokenizer_fingerprint`), each possibly None.
        type_meta: Type-specific reconstruction metadata (sidecar only, excluded from the
            manifest record).
    """

    id: str
    encoding: str
    type: str
    artifact_class: str = "opaque"
    source: str | None = None
    fit_digest: str | None = None
    provenance: dict = field(default_factory=dict)
    type_meta: dict = field(default_factory=dict)

    def manifest_entry(self) -> dict:
        """The record as it appears in `spipe.json` (without `type_meta`)."""
        return {
            "id": self.id,
            "encoding": self.encoding,
            "type": self.type,
            "artifact_class": self.artifact_class,
            "source": self.source,
            "fit_digest": self.fit_digest,
            "provenance": {
                "backend_spec_hash": self.provenance.get("backend_spec_hash"),
                "model_fingerprint": self.provenance.get("model_fingerprint"),
                "tokenizer_fingerprint": self.provenance.get("tokenizer_fingerprint"),
            },
        }

    def sidecar_entry(self) -> dict:
        """The record as written to the artifact's `artifact.json` sidecar."""
        return {**self.manifest_entry(), "type_meta": self.type_meta}

    @classmethod
    def from_mapping(cls, data: Mapping) -> "ArtifactRecord":
        return cls(
            id=data["id"],
            encoding=data["encoding"],
            type=data["type"],
            artifact_class=data.get("artifact_class", "opaque"),
            source=data.get("source"),
            fit_digest=data.get("fit_digest"),
            provenance=dict(data.get("provenance") or {}),
            type_meta=dict(data.get("type_meta") or {}),
        )

artifact_class = 'opaque' class-attribute instance-attribute

encoding instance-attribute

fit_digest = None class-attribute instance-attribute

id instance-attribute

provenance = field(default_factory=dict) class-attribute instance-attribute

source = None class-attribute instance-attribute

type instance-attribute

type_meta = field(default_factory=dict) class-attribute instance-attribute

from_mapping(data) classmethod

Source code in steerability/spipe/store.py

@classmethod
def from_mapping(cls, data: Mapping) -> "ArtifactRecord":
    return cls(
        id=data["id"],
        encoding=data["encoding"],
        type=data["type"],
        artifact_class=data.get("artifact_class", "opaque"),
        source=data.get("source"),
        fit_digest=data.get("fit_digest"),
        provenance=dict(data.get("provenance") or {}),
        type_meta=dict(data.get("type_meta") or {}),
    )

manifest_entry()

The record as it appears in spipe.json (without type_meta).

Source code in steerability/spipe/store.py

def manifest_entry(self) -> dict:
    """The record as it appears in `spipe.json` (without `type_meta`)."""
    return {
        "id": self.id,
        "encoding": self.encoding,
        "type": self.type,
        "artifact_class": self.artifact_class,
        "source": self.source,
        "fit_digest": self.fit_digest,
        "provenance": {
            "backend_spec_hash": self.provenance.get("backend_spec_hash"),
            "model_fingerprint": self.provenance.get("model_fingerprint"),
            "tokenizer_fingerprint": self.provenance.get("tokenizer_fingerprint"),
        },
    }

sidecar_entry()

The record as written to the artifact's artifact.json sidecar.

Source code in steerability/spipe/store.py

def sidecar_entry(self) -> dict:
    """The record as written to the artifact's `artifact.json` sidecar."""
    return {**self.manifest_entry(), "type_meta": self.type_meta}

ArtifactStore

Directory-backed store of content-addressed artifacts.

Parameters:

Name	Type	Description	Default
root	str | Path	The store directory (artifacts/ inside a spipe, or any external directory for thin exports). Created on first write.	required
resolver	Callable[[str], Path] | None	Optional callable mapping an artifact id to the directory holding that artifact, used when artifacts live outside root (a thin export's external store).	None

Source code in steerability/spipe/store.py

class ArtifactStore:
    """Directory-backed store of content-addressed artifacts.

    Args:
        root: The store directory (`artifacts/` inside a spipe, or any external directory for
            thin exports). Created on first write.
        resolver: Optional callable mapping an artifact id to the directory holding that
            artifact, used when artifacts live outside `root` (a thin export's external
            store).
    """

    def __init__(self, root: str | Path, resolver: Callable[[str], Path] | None = None):
        self.root = Path(root)
        self._resolver = resolver

    def _dir_for(self, artifact_id: str) -> Path:
        local = self.root / _dir_name(artifact_id)
        if local.exists() or self._resolver is None:
            return local
        resolved = Path(self._resolver(artifact_id))
        return resolved if resolved.exists() else local

    def has(self, artifact_id: str) -> bool:
        """Whether the store holds `artifact_id`."""
        return (self._dir_for(artifact_id) / SIDECAR_FILE).exists()

    def ids(self) -> list[str]:
        """Sorted ids of every artifact under `root`."""
        if not self.root.is_dir():
            return []
        return sorted(
            _artifact_id_from_dir_name(child.name)
            for child in self.root.iterdir()
            if child.is_dir() and (child / SIDECAR_FILE).exists()
        )

    def _write_sidecar(self, directory: Path, record: ArtifactRecord) -> None:
        with open(directory / SIDECAR_FILE, "w", encoding="utf-8") as handle:
            json.dump(record.sidecar_entry(), handle, sort_keys=True, indent=2)

    def put_tensors(self, tensors: Mapping[str, torch.Tensor], record_fields: dict) -> ArtifactRecord:
        """Write a tensor payload as a `"tensors"` artifact, idempotently.

        Args:
            tensors: Mapping from tensor name to tensor.
            record_fields: Record fields other than `id` and `encoding` (`type`,
                `artifact_class`, `source`, `fit_digest`, `provenance`, `type_meta`).

        Returns:
            The artifact record.
        """
        artifact_id, data = tensors_payload(tensors)
        record = ArtifactRecord(id=artifact_id, encoding="tensors", **record_fields)
        directory = self.root / _dir_name(artifact_id)
        if not (directory / TENSOR_FILE).exists():
            directory.mkdir(parents=True, exist_ok=True)
            (directory / TENSOR_FILE).write_bytes(data)
            self._write_sidecar(directory, record)
        return record

    def put_tree(self, source: str | Path, record_fields: dict) -> ArtifactRecord:
        """Copy a directory as a `"tree"` artifact, idempotently.

        Args:
            source: Directory whose contents become `payload/`. Symlinks are rejected.
            record_fields: Record fields other than `id` and `encoding`.

        Returns:
            The artifact record.

        Raises:
            SpipeSaveError: If `source` is missing, empty, or contains a symlink.
        """
        source = Path(source)
        artifact_id = tree_id_for(source)
        record = ArtifactRecord(id=artifact_id, encoding="tree", **record_fields)
        directory = self.root / _dir_name(artifact_id)
        if not (directory / PAYLOAD_DIR).exists():
            directory.mkdir(parents=True, exist_ok=True)
            shutil.copytree(source, directory / PAYLOAD_DIR, symlinks=False)
            self._write_sidecar(directory, record)
        return record

    def record_for(self, artifact_id: str) -> ArtifactRecord:
        """The sidecar record of a stored artifact.

        Raises:
            SpipeIntegrityError: If the artifact is absent or the sidecar's `id` does not
                match the directory name.
        """
        directory = self._dir_for(artifact_id)
        sidecar = directory / SIDECAR_FILE
        if not sidecar.exists():
            raise SpipeIntegrityError(
                f"Artifact {artifact_id} is not in the store at {self.root}"
                + (" (no external resolver matched)" if self._resolver else
                   "; for a thin bundle, pass artifact_store= to load()")
                + "."
            )
        record = ArtifactRecord.from_mapping(json.loads(sidecar.read_text(encoding="utf-8")))
        if record.id != artifact_id:
            raise SpipeIntegrityError(
                f"Artifact sidecar at {directory} records id {record.id} but the directory "
                f"is named for {artifact_id}."
            )
        return record

    def verify(self, artifact_id: str) -> None:
        """Verify a stored artifact's bytes against its content id.

        Raises:
            SpipeIntegrityError: If the artifact is absent, the sidecar id mismatches, or the
                recomputed content hash differs from `artifact_id`.
        """
        record = self.record_for(artifact_id)
        directory = self._dir_for(artifact_id)
        if record.encoding == "tensors":
            data = (directory / TENSOR_FILE).read_bytes()
            actual = "sha256:" + hashlib.sha256(data).hexdigest()
        else:
            actual = tree_id_for(directory / PAYLOAD_DIR)
        if actual != artifact_id:
            raise SpipeIntegrityError(
                f"Artifact {artifact_id} failed integrity verification (content hashes to "
                f"{actual})."
            )

    def load_tensors(self, artifact_id: str) -> dict[str, torch.Tensor]:
        """Load and verify a `"tensors"` artifact.

        Returns:
            Mapping from tensor name to float32 CPU tensor.
        """
        import safetensors.torch

        self.verify(artifact_id)
        directory = self._dir_for(artifact_id)
        return safetensors.torch.load_file(str(directory / TENSOR_FILE))

    def payload_path(self, artifact_id: str) -> Path:
        """The verified `payload/` path of a `"tree"` artifact."""
        self.verify(artifact_id)
        return self._dir_for(artifact_id) / PAYLOAD_DIR

    def size_of(self, artifact_id: str) -> int:
        """Total on-disk bytes of an artifact's content (sidecar excluded)."""
        directory = self._dir_for(artifact_id)
        record = self.record_for(artifact_id)
        if record.encoding == "tensors":
            return (directory / TENSOR_FILE).stat().st_size
        return sum(p.stat().st_size for p in (directory / PAYLOAD_DIR).rglob("*") if p.is_file())

    def copy_into(self, dest_root: str | Path, artifact_ids: list[str]) -> None:
        """Copy the named artifacts into another store directory (fat export)."""
        dest_root = Path(dest_root)
        for artifact_id in artifact_ids:
            source = self._dir_for(artifact_id)
            dest = dest_root / _dir_name(artifact_id)
            if not dest.exists():
                dest.mkdir(parents=True, exist_ok=True)
                shutil.copytree(source, dest, symlinks=False, dirs_exist_ok=True)

root = Path(root) instance-attribute

copy_into(dest_root, artifact_ids)

Copy the named artifacts into another store directory (fat export).

Source code in steerability/spipe/store.py

def copy_into(self, dest_root: str | Path, artifact_ids: list[str]) -> None:
    """Copy the named artifacts into another store directory (fat export)."""
    dest_root = Path(dest_root)
    for artifact_id in artifact_ids:
        source = self._dir_for(artifact_id)
        dest = dest_root / _dir_name(artifact_id)
        if not dest.exists():
            dest.mkdir(parents=True, exist_ok=True)
            shutil.copytree(source, dest, symlinks=False, dirs_exist_ok=True)

has(artifact_id)

Whether the store holds artifact_id.

Source code in steerability/spipe/store.py

def has(self, artifact_id: str) -> bool:
    """Whether the store holds `artifact_id`."""
    return (self._dir_for(artifact_id) / SIDECAR_FILE).exists()

ids()

Sorted ids of every artifact under root.

Source code in steerability/spipe/store.py

def ids(self) -> list[str]:
    """Sorted ids of every artifact under `root`."""
    if not self.root.is_dir():
        return []
    return sorted(
        _artifact_id_from_dir_name(child.name)
        for child in self.root.iterdir()
        if child.is_dir() and (child / SIDECAR_FILE).exists()
    )

load_tensors(artifact_id)

Load and verify a "tensors" artifact.

Returns:

Type	Description
dict[str, Tensor]	Mapping from tensor name to float32 CPU tensor.

Source code in steerability/spipe/store.py

def load_tensors(self, artifact_id: str) -> dict[str, torch.Tensor]:
    """Load and verify a `"tensors"` artifact.

    Returns:
        Mapping from tensor name to float32 CPU tensor.
    """
    import safetensors.torch

    self.verify(artifact_id)
    directory = self._dir_for(artifact_id)
    return safetensors.torch.load_file(str(directory / TENSOR_FILE))

payload_path(artifact_id)

The verified payload/ path of a "tree" artifact.

Source code in steerability/spipe/store.py

def payload_path(self, artifact_id: str) -> Path:
    """The verified `payload/` path of a `"tree"` artifact."""
    self.verify(artifact_id)
    return self._dir_for(artifact_id) / PAYLOAD_DIR

put_tensors(tensors, record_fields)

Write a tensor payload as a "tensors" artifact, idempotently.

Parameters:

Name	Type	Description	Default
tensors	Mapping[str, Tensor]	Mapping from tensor name to tensor.	required
record_fields	dict	Record fields other than id and encoding (type, artifact_class, source, fit_digest, provenance, type_meta).	required

Returns:

Type	Description
ArtifactRecord	The artifact record.

Source code in steerability/spipe/store.py

def put_tensors(self, tensors: Mapping[str, torch.Tensor], record_fields: dict) -> ArtifactRecord:
    """Write a tensor payload as a `"tensors"` artifact, idempotently.

    Args:
        tensors: Mapping from tensor name to tensor.
        record_fields: Record fields other than `id` and `encoding` (`type`,
            `artifact_class`, `source`, `fit_digest`, `provenance`, `type_meta`).

    Returns:
        The artifact record.
    """
    artifact_id, data = tensors_payload(tensors)
    record = ArtifactRecord(id=artifact_id, encoding="tensors", **record_fields)
    directory = self.root / _dir_name(artifact_id)
    if not (directory / TENSOR_FILE).exists():
        directory.mkdir(parents=True, exist_ok=True)
        (directory / TENSOR_FILE).write_bytes(data)
        self._write_sidecar(directory, record)
    return record

put_tree(source, record_fields)

Copy a directory as a "tree" artifact, idempotently.

Parameters:

Name	Type	Description	Default
source	str | Path	Directory whose contents become payload/. Symlinks are rejected.	required
record_fields	dict	Record fields other than id and encoding.	required

Returns:

Type	Description
ArtifactRecord	The artifact record.

Raises:

Type	Description
SpipeSaveError	If source is missing, empty, or contains a symlink.

Source code in steerability/spipe/store.py

def put_tree(self, source: str | Path, record_fields: dict) -> ArtifactRecord:
    """Copy a directory as a `"tree"` artifact, idempotently.

    Args:
        source: Directory whose contents become `payload/`. Symlinks are rejected.
        record_fields: Record fields other than `id` and `encoding`.

    Returns:
        The artifact record.

    Raises:
        SpipeSaveError: If `source` is missing, empty, or contains a symlink.
    """
    source = Path(source)
    artifact_id = tree_id_for(source)
    record = ArtifactRecord(id=artifact_id, encoding="tree", **record_fields)
    directory = self.root / _dir_name(artifact_id)
    if not (directory / PAYLOAD_DIR).exists():
        directory.mkdir(parents=True, exist_ok=True)
        shutil.copytree(source, directory / PAYLOAD_DIR, symlinks=False)
        self._write_sidecar(directory, record)
    return record

record_for(artifact_id)

The sidecar record of a stored artifact.

Raises:

Type	Description
SpipeIntegrityError	If the artifact is absent or the sidecar's id does not match the directory name.

Source code in steerability/spipe/store.py

def record_for(self, artifact_id: str) -> ArtifactRecord:
    """The sidecar record of a stored artifact.

    Raises:
        SpipeIntegrityError: If the artifact is absent or the sidecar's `id` does not
            match the directory name.
    """
    directory = self._dir_for(artifact_id)
    sidecar = directory / SIDECAR_FILE
    if not sidecar.exists():
        raise SpipeIntegrityError(
            f"Artifact {artifact_id} is not in the store at {self.root}"
            + (" (no external resolver matched)" if self._resolver else
               "; for a thin bundle, pass artifact_store= to load()")
            + "."
        )
    record = ArtifactRecord.from_mapping(json.loads(sidecar.read_text(encoding="utf-8")))
    if record.id != artifact_id:
        raise SpipeIntegrityError(
            f"Artifact sidecar at {directory} records id {record.id} but the directory "
            f"is named for {artifact_id}."
        )
    return record

size_of(artifact_id)

Total on-disk bytes of an artifact's content (sidecar excluded).

Source code in steerability/spipe/store.py

def size_of(self, artifact_id: str) -> int:
    """Total on-disk bytes of an artifact's content (sidecar excluded)."""
    directory = self._dir_for(artifact_id)
    record = self.record_for(artifact_id)
    if record.encoding == "tensors":
        return (directory / TENSOR_FILE).stat().st_size
    return sum(p.stat().st_size for p in (directory / PAYLOAD_DIR).rglob("*") if p.is_file())

verify(artifact_id)

Verify a stored artifact's bytes against its content id.

Raises:

Type	Description
SpipeIntegrityError	If the artifact is absent, the sidecar id mismatches, or the recomputed content hash differs from artifact_id.

Source code in steerability/spipe/store.py

def verify(self, artifact_id: str) -> None:
    """Verify a stored artifact's bytes against its content id.

    Raises:
        SpipeIntegrityError: If the artifact is absent, the sidecar id mismatches, or the
            recomputed content hash differs from `artifact_id`.
    """
    record = self.record_for(artifact_id)
    directory = self._dir_for(artifact_id)
    if record.encoding == "tensors":
        data = (directory / TENSOR_FILE).read_bytes()
        actual = "sha256:" + hashlib.sha256(data).hexdigest()
    else:
        actual = tree_id_for(directory / PAYLOAD_DIR)
    if actual != artifact_id:
        raise SpipeIntegrityError(
            f"Artifact {artifact_id} failed integrity verification (content hashes to "
            f"{actual})."
        )

save_object_tree(save_fn, store, record_fields)

Save an object into the store as a "tree" artifact via its own save(path) method.

Parameters:

Name	Type	Description	Default
save_fn	Callable[[Path], Any]	Callable writing the object into a directory (or file inside it).	required
store	ArtifactStore	The destination store.	required
record_fields	dict	Record fields other than id and encoding.	required

Returns:

Type	Description
ArtifactRecord	The artifact record.

Source code in steerability/spipe/store.py

def save_object_tree(save_fn: Callable[[Path], Any], store: ArtifactStore, record_fields: dict) -> ArtifactRecord:
    """Save an object into the store as a `"tree"` artifact via its own `save(path)` method.

    Args:
        save_fn: Callable writing the object into a directory (or file inside it).
        store: The destination store.
        record_fields: Record fields other than `id` and `encoding`.

    Returns:
        The artifact record.
    """
    import tempfile

    with tempfile.TemporaryDirectory(prefix="spipe-artifact-") as tmp:
        save_fn(Path(tmp))
        return store.put_tree(tmp, record_fields)

tensors_payload(tensors)

The content id and serialized safetensors bytes of a tensor payload.

Uses the id algorithm of artifact_id_for (float32, contiguous, CPU, sorted names, SHA-256 over the safetensors serialization), which makes tensor artifact directories byte-compatible with the vLLM-Hook plugin registry.

Parameters:

Name	Type	Description	Default
tensors	Mapping[str, Tensor]	Mapping from tensor name to tensor.	required

Returns:

Type	Description
tuple[str, bytes]	The sha256:<hex> id and the serialized bytes.

Source code in steerability/spipe/store.py

def tensors_payload(tensors: Mapping[str, torch.Tensor]) -> tuple[str, bytes]:
    """The content id and serialized safetensors bytes of a tensor payload.

    Uses the id algorithm of `artifact_id_for` (float32, contiguous, CPU, sorted names,
    SHA-256 over the safetensors serialization), which makes tensor artifact directories
    byte-compatible with the vLLM-Hook plugin registry.

    Args:
        tensors: Mapping from tensor name to tensor.

    Returns:
        The `sha256:<hex>` id and the serialized bytes.
    """
    import safetensors.torch

    from steerability.algorithms.state_control.common.lowering import artifact_id_for

    artifact_id, prepared = artifact_id_for(tensors)
    data = safetensors.torch.save({name: prepared[name] for name in sorted(prepared)})
    return artifact_id, data

tree_id_for(root)

The content id of a directory tree.

The id is "sha256:" + sha256 over the concatenation of f"{posix_relpath}\n{sha256_of_file_hex}\n" for every file, sorted by posix relative path. Symlinks are rejected.

Parameters:

Name	Type	Description	Default
root	Path	Directory to hash.	required

Returns:

Type	Description
str	The sha256:<hex> id.

Raises:

Type	Description
SpipeSaveError	If root is not a directory, is empty, or contains a symlink.

Source code in steerability/spipe/store.py

def tree_id_for(root: Path) -> str:
    """The content id of a directory tree.

    The id is `"sha256:" + sha256` over the concatenation of
    `f"{posix_relpath}\\n{sha256_of_file_hex}\\n"` for every file, sorted by posix relative
    path. Symlinks are rejected.

    Args:
        root: Directory to hash.

    Returns:
        The `sha256:<hex>` id.

    Raises:
        SpipeSaveError: If `root` is not a directory, is empty, or contains a symlink.
    """
    root = Path(root)
    if not root.is_dir():
        raise SpipeSaveError(f"Tree artifact source {root} is not a directory.")
    entries: list[tuple[str, str]] = []
    for path in sorted(root.rglob("*")):
        if path.is_symlink():
            raise SpipeSaveError(f"Tree artifact source contains a symlink: {path}.")
        if path.is_file():
            entries.append((path.relative_to(root).as_posix(), _file_sha256(path)))
    if not entries:
        raise SpipeSaveError(f"Tree artifact source {root} contains no files.")
    digest = hashlib.sha256()
    for relpath, file_hash in entries:
        digest.update(f"{relpath}\n{file_hash}\n".encode("utf-8"))
    return "sha256:" + digest.hexdigest()