diff --git a/packages/kumo-api/kumoapi/plugin.py b/packages/kumo-api/kumoapi/plugin.py
index b649e302b..018755408 100644
--- a/packages/kumo-api/kumoapi/plugin.py
+++ b/packages/kumo-api/kumoapi/plugin.py
@@ -21,10 +21,12 @@
 :class:`~kumoai_studio.backend.LocalBackend`.
 """
 
+import re
 from dataclasses import dataclass, field
 from typing import (
     Any,
     Callable,
+    Dict,
     List,
     Mapping,
     Optional,
@@ -42,10 +44,14 @@
 from kumoapi.typing import Dtype
 
 __all__ = [
+    'BUNDLE_ID_KEY',
+    'ENGINE_MARKER',
+    'REGISTRY_KEY_PREFIX',
     'ArtifactLocation',
     'ArtifactSink',
     'FileStoreSink',
     'IngestResult',
+    'ModelAliasingSink',
     'ModelFetchingSink',
     'ModelListingSink',
     'ModelPublishingSink',
@@ -56,8 +62,20 @@
     'SourceBackend',
     'SourceConnector',
     'SourceCredentials',
+    'check_model_alias',
+    'registry_lineage',
 ]
 
+#: Prefix of the keys an engine export records in registry metadata.
+REGISTRY_KEY_PREFIX = 'kumo_'
+#: Registry metadata key marking a version as a Kumo engine bundle.
+ENGINE_MARKER = 'kumo_engine'
+#: Registry metadata key holding the version's bundle id; the bundle's own
+#: provenance.json holds its lineage.
+BUNDLE_ID_KEY = 'kumo_bundle_id'
+_ALIAS = re.compile(r'[A-Za-z_][A-Za-z0-9_]{0,254}')
+_RESERVED_ALIAS = re.compile(r'DEFAULT|FIRST|LAST|LATEST|V[0-9]+', re.I)
+
 
 @dataclass(frozen=True)
 class SourceCredentials:
@@ -149,13 +167,16 @@ class ModelRegistryLocation:
     Args:
         model_name: Registry-local model name.
         version: Exact version to write or read. ``None`` lets the registry
-            assign the next one on export, and selects the latest on import.
-            An assigned version comes back on the export result.
+            assign the next one on export; on import it selects Snowflake's
+            default or Databricks' latest Kumo version.
         comment: Optional description recorded on the version.
+        alias: Set on the new version after export; on import, the version
+            it points at. Exclusive with :obj:`version`.
     """
     model_name: str
     version: Optional[str] = None
     comment: Optional[str] = None
+    alias: Optional[str] = None
 
     def __post_init__(self) -> None:
         if (not isinstance(self.model_name, str) or not self.model_name
@@ -166,6 +187,10 @@ def __post_init__(self) -> None:
             raise ValueError(
                 'version must be a non-empty version, or None to let the '
                 'registry choose')
+        if self.alias is not None and (not self.alias
+                                       or self.version is not None):
+            raise ValueError('alias must be a non-empty alias, and is '
+                             'mutually exclusive with version')
 
 
 @dataclass(frozen=True)
@@ -705,7 +730,8 @@ def fetch_model(
             local_dir: Directory to place the bundle in. Created if absent.
             path: Source for a path-addressed target.
             location: A registry location. A version of ``None`` selects the
-                latest, which is resolved here and reported back.
+                registry's choice (see :class:`ModelRegistryLocation`), which
+                is resolved here and reported back.
             project_id: Authorizing project, where one applies.
 
         Returns:
@@ -715,6 +741,44 @@ def fetch_model(
         ...
 
 
+@runtime_checkable
+class ModelAliasingSink(ArtifactSink, Protocol):
+    r"""A registry sink that sets aliases and fetches by
+    :attr:`ModelRegistryLocation.alias`."""
+    def set_model_alias(
+        self,
+        connector: SourceConnector,
+        location: ModelRegistryLocation,
+        alias: str,
+    ) -> None:
+        r"""Points :obj:`alias` at ``location.version``, moving it if held."""
+        ...
+
+
+def check_model_alias(alias: str) -> None:
+    r"""Raises :class:`ValueError` unless every model registry can set
+    :obj:`alias`: a letter or underscore, then letters, digits or underscores
+    (at most 255), and not ``DEFAULT``, ``FIRST``, ``LAST``, ``LATEST`` or
+    ``v<digits>`` in any case."""
+    if (not isinstance(alias, str) or not _ALIAS.fullmatch(alias)
+            or _RESERVED_ALIAS.fullmatch(alias)):
+        raise ValueError(
+            f'{alias!r} cannot be set as a model alias: use letters, digits '
+            f'and underscores, starting with a letter or underscore, and '
+            f'avoid the reserved DEFAULT, FIRST, LAST, LATEST and v<digits> '
+            f'(for DEFAULT use ALTER MODEL ... SET DEFAULT_VERSION).')
+
+
+def registry_lineage(recorded: Mapping[str, Any]) -> Dict[str, str]:
+    r"""The ``kumo_*`` entries of a version's registry metadata, such as
+    :data:`BUNDLE_ID_KEY`, as strings and without :data:`ENGINE_MARKER`."""
+    return {
+        key: str(value)
+        for key, value in recorded.items()
+        if key.startswith(REGISTRY_KEY_PREFIX) and key != ENGINE_MARKER
+    }
+
+
 @runtime_checkable
 class ModelListingSink(ArtifactSink, Protocol):
     r"""A sink that can enumerate the models published through it.
diff --git a/packages/kumo-api/test/test_plugin_registry.py b/packages/kumo-api/test/test_plugin_registry.py
new file mode 100644
index 000000000..52ca7a69f
--- /dev/null
+++ b/packages/kumo-api/test/test_plugin_registry.py
@@ -0,0 +1,33 @@
+# SPDX-FileCopyrightText: Copyright (c) 2026, NVIDIA CORPORATION & AFFILIATES.
+# All rights reserved.
+# SPDX-License-Identifier: Apache-2.0
+import pytest
+
+from kumoapi.plugin import check_model_alias, registry_lineage
+
+
+@pytest.mark.parametrize('alias', [
+    '', ' ', 'default', 'LAST', 'Latest', 'v3', 'my-alias', '2nd', 'a' * 256,
+    None
+])
+def test_check_model_alias_refuses_what_a_registry_reserves(alias):
+    with pytest.raises(ValueError):
+        check_model_alias(alias)
+
+
+def test_check_model_alias_accepts_a_plain_name():
+    check_model_alias('champion')
+    check_model_alias('_v2_prod')
+
+
+def test_registry_lineage_keeps_kumo_keys_as_strings():
+    recorded = {
+        'kumo_engine': '1',
+        'kumo_bundle_id': 'b',
+        'auc': 0.9,
+        'kumo_seconds': 3
+    }
+    assert registry_lineage(recorded) == {
+        'kumo_bundle_id': 'b',
+        'kumo_seconds': '3'
+    }
diff --git a/packages/kumo-databricks/src/kumo_databricks/registry.py b/packages/kumo-databricks/src/kumo_databricks/registry.py
index d2d88de09..05a50f4a7 100644
--- a/packages/kumo-databricks/src/kumo_databricks/registry.py
+++ b/packages/kumo-databricks/src/kumo_databricks/registry.py
@@ -10,7 +10,6 @@
 import os
 import shutil
 import tempfile
-from dataclasses import replace
 from pathlib import Path
 from typing import Any, Iterator, Mapping, Optional, cast
 from urllib.parse import quote
@@ -20,12 +19,15 @@
 from kumo_databricks.session import _DatabricksSessions, _target_config
 from kumo_databricks.sql import normalize_table_coordinate
 from kumoapi.plugin import (
+    BUNDLE_ID_KEY,
+    ENGINE_MARKER,
     ModelRegistryLocation,
     PublishedModel,
     SourceConnector,
+    check_model_alias,
+    registry_lineage,
 )
 
-_ENGINE_MARKER = 'kumo_engine'
 _MODELS = '/api/2.1/unity-catalog/models'
 _LOG = logging.getLogger(__name__)
 
@@ -56,6 +58,17 @@ def _tags(version: Mapping[str, Any]) -> dict[str, str]:
     return {tag['key']: tag['value'] for tag in version.get('tags', [])}
 
 
+def _alias_path(name: str, alias: str) -> str:
+    return f'{_MODELS}/{quote(name, safe="")}/aliases/{quote(alias, safe="")}'
+
+
+def _version_number(location: ModelRegistryLocation) -> Optional[int]:
+    version = location.version
+    if version is not None and (not version.isdecimal() or int(version) < 1):
+        raise ValueError('Databricks model versions are positive integers.')
+    return None if version is None else int(version)
+
+
 def _published(
     name: str,
     version: Mapping[str, Any],
@@ -133,7 +146,9 @@ def publish_bundle(
     verify_model_bundle(root)
     _validate_catalog(dict(metadata=dict(metadata), metrics=dict(metrics)))
     catalog_name, schema, model = name.split('.')
-    tags = {_ENGINE_MARKER: '1'}
+    tags = {ENGINE_MARKER: '1'}
+    if metadata.get(BUNDLE_ID_KEY):
+        tags[BUNDLE_ID_KEY] = str(metadata[BUNDLE_ID_KEY])
     try:
         client.registered_models.create(catalog_name=catalog_name,
                                         schema_name=schema, name=model)
@@ -200,7 +215,7 @@ def _bundle_metadata(bundle: Path) -> tuple[dict[str, str], dict[str, float]]:
 def _download(client: WorkspaceClient, remote: str, destination: Path) -> None:
     for entry in client.files.list_directory_contents(remote):
         # A server-provided path must stay an immediate child of this directory.
-        path = entry.path or ''
+        path = (entry.path or '').rstrip('/')
         prefix = remote + '/'
         child = path[len(prefix):] if path.startswith(prefix) else ''
         if not child or '/' in child or child in ('.', '..') or '\\' in child:
@@ -219,13 +234,18 @@ def _download(client: WorkspaceClient, remote: str, destination: Path) -> None:
 
 
 def fetch_bundle(client: WorkspaceClient, name: str, local_dir: Path, *,
-                 version: Optional[str] = None) -> PublishedModel:
+                 version: Optional[str] = None,
+                 alias: Optional[str] = None) -> PublishedModel:
     from kumo_core.artifacts import verify_model_bundle
 
-    if version is None:
+    if alias is not None:
+        aliased = cast(dict[str, Any],
+                       client.api_client.do('GET', _alias_path(name, alias)))
+        model_version = _get_version(client, name, aliased['version'])
+    elif version is None:
         versions = [
             v for v in _versions(client, name) if v.get('status') == 'READY'
-            and _tags(v).get(_ENGINE_MARKER) == '1'
+            and _tags(v).get(ENGINE_MARKER) == '1'
         ]
         if not versions:
             raise ValueError('No ready Kumo model version in this registry.')
@@ -233,7 +253,7 @@ def fetch_bundle(client: WorkspaceClient, name: str, local_dir: Path, *,
     else:
         model_version = _get_version(client, name, version)
     if model_version.get('status') != 'READY' or _tags(model_version).get(
-            _ENGINE_MARKER) != '1':
+            ENGINE_MARKER) != '1':
         raise ValueError('This version is not a ready Kumo model bundle.')
     destination = local_dir
     destination.parent.mkdir(parents=True, exist_ok=True)
@@ -251,9 +271,12 @@ def fetch_bundle(client: WorkspaceClient, name: str, local_dir: Path, *,
                 'The registered model must contain one Kumo bundle.')
         bundle = bundles[0]
         verify_model_bundle(bundle)
-        published = _published(name, model_version)
         metadata, metrics = _bundle_metadata(bundle)
-        published = replace(published, metadata=metadata, metrics=metrics)
+        published = _published(
+            name, model_version, metadata={
+                **metadata,
+                **registry_lineage(_tags(model_version))
+            }, metrics=metrics)
         backup = Path(tmp) / 'previous'
         if destination.exists():
             os.replace(destination, backup)
@@ -277,8 +300,10 @@ def list_models(client: WorkspaceClient, catalog: str,
             continue
         for version in _versions(client, name):
             if version.get('status') == 'READY' and _tags(version).get(
-                    _ENGINE_MARKER) == '1':
-                found.append(_published(name, version))
+                    ENGINE_MARKER) == '1':
+                found.append(
+                    _published(name, version,
+                               metadata=registry_lineage(_tags(version))))
     return sorted(found, key=lambda item: (item.name, int(item.version)))
 
 
@@ -321,15 +346,26 @@ def fetch_model(
         location: Optional[ModelRegistryLocation] = None,
         project_id: Optional[str] = None,
     ) -> PublishedModel:
-        """Download the named version, or the latest ready Kumo version."""
+        """Download the named or aliased version, else the latest Kumo one."""
         del project_id
         name = _address(connector, path, location)
         assert location is not None
-        if location.version is not None and (not location.version.isdecimal()
-                                             or int(location.version) < 1):
-            raise ValueError('Databricks model versions are positive integers.')
+        _version_number(location)
         return fetch_bundle(self._client(connector), name,
-                            Path(local_dir).resolve(), version=location.version)
+                            Path(local_dir).resolve(), version=location.version,
+                            alias=getattr(location, 'alias', None))
+
+    def set_model_alias(self, connector: SourceConnector,
+                        location: ModelRegistryLocation, alias: str) -> None:
+        """Point an alias at a version; UC moves it from any other."""
+        name = _address(connector, None, location)
+        number = _version_number(location)
+        if number is None:
+            raise ValueError('Setting an alias needs a ModelRegistryLocation '
+                             'naming a version.')
+        check_model_alias(alias)
+        self._client(connector).api_client.do('PUT', _alias_path(name, alias),
+                                              body=dict(version_num=number))
 
     def list_models(self, connector: SourceConnector, *,
                     project_id: Optional[str] = None) -> list[PublishedModel]:
diff --git a/packages/kumo-databricks/tests/test_registry.py b/packages/kumo-databricks/tests/test_registry.py
index 050fcf18e..4e4c8cee4 100644
--- a/packages/kumo-databricks/tests/test_registry.py
+++ b/packages/kumo-databricks/tests/test_registry.py
@@ -15,6 +15,8 @@
 from kumo_databricks.sink import DatabricksSinkBackend
 from kumoapi.data_source import DatabricksConnectorResourceConfig
 from kumoapi.plugin import (
+    ENGINE_MARKER,
+    ModelAliasingSink,
     ModelFetchingSink,
     ModelListingSink,
     ModelPublishingSink,
@@ -61,6 +63,7 @@ def test_sink_implements_registry_protocols():
     assert isinstance(sink, ModelPublishingSink)
     assert isinstance(sink, ModelFetchingSink)
     assert isinstance(sink, ModelListingSink)
+    assert isinstance(sink, ModelAliasingSink)
 
 
 def test_publish_uses_connector_client_and_reports_assigned_version(
@@ -141,7 +144,7 @@ def test_invalid_fetch_version_fails_before_authentication(
 
 
 def version(number, kumo=True, status='READY'):
-    tags = {registry._ENGINE_MARKER: '1'} if kumo else {}
+    tags = {ENGINE_MARKER: '1'} if kumo else {}
     return dict(version=number, status=status,
                 tags=[dict(key=k, value=v) for k, v in tags.items()])
 
@@ -155,6 +158,7 @@ def __init__(self):
             upload=self.upload, list_directory_contents=self.list_directory,
             download=self.download)
         self.objects = {}
+        self.aliases = {}
         self.calls = []
         self.deleted = []
         self.versions = [
@@ -173,6 +177,12 @@ def delete(self, name, number):
 
     def do(self, method, path, query=None, body=None):
         self.calls.append((method, path, query, body))
+        if '/aliases/' in path:
+            alias = path.rsplit('/', 1)[-1]
+            if method == 'PUT':
+                self.aliases[alias] = body['version_num']
+                return {}
+            return dict(version=self.aliases[alias])
         if method == 'POST':
             assert all(len(tag['value']) <= 256 for tag in body['tags'])
             self.versions.append(
@@ -207,8 +217,11 @@ def list_directory(self, path):
             if key.startswith(path + '/'):
                 relative = key[len(path) + 1:]
                 child = relative.split('/')[0]
-                children[child] = SimpleNamespace(path=path + '/' + child,
-                                                  is_directory='/' in relative)
+                directory = '/' in relative
+                # The Files API lists directories with a trailing slash.
+                children[child] = SimpleNamespace(
+                    path=f"{path}/{child}{'/' if directory else ''}",
+                    is_directory=directory)
         return children.values()
 
     def download(self, path):
@@ -485,3 +498,49 @@ def test_invalid_bundle_metadata_keeps_destination(client, bundle, tmp_path,
         registry.fetch_bundle(client, name='main.analytics.churn',
                               local_dir=destination, version='7')
     assert (destination / 'weights.bin').read_bytes() == b'previous'
+
+
+def test_bundle_id_is_tagged_listed_and_fetched_by_version_and_alias(
+        connector, client, bundle, tmp_path, monkeypatch):
+    lineage = dict(kumo_bundle_id='ab' * 32)
+    sink = DatabricksSinkBackend()
+    monkeypatch.setattr(sink, '_client', lambda _: client)
+    published = sink.publish_model(
+        connector, str(bundle), location=ModelRegistryLocation('churn'),
+        metadata=dict(spec_version='1', kumo_engine='0', kumo_run_id='run-1',
+                      **lineage))
+    assert registry._tags(client.versions[-1]) == {
+        'kumo_engine': '1',
+        **lineage
+    }
+    listed = registry.list_models(client, catalog='main', schema='analytics')
+    assert next(m for m in listed if m.version == '7').metadata == lineage
+
+    sink.set_model_alias(
+        connector, ModelRegistryLocation('churn', version=published.version),
+        'prod')
+    assert client.calls[-1] == (
+        'PUT',
+        '/api/2.1/unity-catalog/models/main.analytics.churn/aliases/prod', None,
+        dict(version_num=7))
+    for location in (ModelRegistryLocation('churn', version='7'),
+                     ModelRegistryLocation('churn', alias='prod')):
+        fetched = sink.fetch_model(connector, str(tmp_path / 'restored'),
+                                   location=location)
+        assert fetched.version == '7'
+        assert fetched.metadata == {
+            'spec_version': '1',
+            'predictive_query': 'PREDICT COUNT(*)',
+            'snapshot_id': 'snapshot',
+            **lineage
+        }
+
+
+def test_set_alias_needs_a_positive_version(connector, monkeypatch):
+    sink = DatabricksSinkBackend()
+    monkeypatch.setattr(sink, '_client',
+                        lambda _: pytest.fail('resolved identity'))
+    with pytest.raises(ValueError, match='positive integer'):
+        sink.set_model_alias(connector,
+                             ModelRegistryLocation('churn', version='0'),
+                             'prod')
diff --git a/packages/kumo-databricks/tests/test_uc_registry.py b/packages/kumo-databricks/tests/test_uc_registry.py
index 8866cf80f..8acca5647 100644
--- a/packages/kumo-databricks/tests/test_uc_registry.py
+++ b/packages/kumo-databricks/tests/test_uc_registry.py
@@ -76,7 +76,7 @@ def handle_request(self):
                 return self.send(dict(registered_models=[model]))
             if path.endswith('/models/versions') and self.command == 'POST':
                 assert payload['source'].startswith('file:')
-                assert payload['tags'] == [{'key': 'kumo_engine', 'value': '1'}]
+                assert {'key': 'kumo_engine', 'value': '1'} in payload['tags']
                 versions[identity] = dict(version=1,
                                           status='PENDING_REGISTRATION',
                                           tags=payload['tags'])
diff --git a/packages/kumo-engine/docs/mlops.md b/packages/kumo-engine/docs/mlops.md
index 8a12d1473..3d7df6802 100644
--- a/packages/kumo-engine/docs/mlops.md
+++ b/packages/kumo-engine/docs/mlops.md
@@ -62,6 +62,32 @@ tuned.scoring_defaults   # ScoringDefaults(binary_classification_threshold=0.37,
 
 `with_scoring_defaults` returns a `TrainingResult` for a copy of the bundle with its own id. The copy's manifest embeds the original's in `parents.derived_from`. The threshold applies to binary and multilabel classification, `num_classes_to_return` to multiclass, multilabel and link prediction. Explicit `predict` arguments and a query's `TOP k` win; `ScoringDefaults()` removes the defaults. `Trainer.export(config)` publishes a fitted or loaded bundle like `fit(export_config=...)`.
 
+### Registering models
+
+With `registry=ModelOutputConfig(connector, location=...)`, every fit in the block registers its model in the connector's registry (Snowflake Model Registry, or Unity Catalog through a `DatabricksConnector`) after it is stored and tracked:
+
+```python
+from kumo.engine import ModelOutputConfig, ModelRegistryLocation
+
+cfg = kumo.MLOps("s3://bucket/kumo", registry=ModelOutputConfig(
+    registry, location=ModelRegistryLocation("CHURN_GNN", alias="challenger")))
+with kumo.mlops(cfg):
+    result = trainer.fit(training_table, snapshot=snapshot)
+result.registered_model   # PublishedModel(name='CHURN_GNN', version='V3', uri=...)
+
+# Promote: the alias moves from any version that held it.
+kumo.set_model_alias(registry, ModelRegistryLocation("CHURN_GNN", version="V3"), "prod")
+
+# Scoring jobs load whichever version "prod" points at.
+trainer = kumo.Trainer.load_from_connector(registry, location=ModelRegistryLocation(
+    "CHURN_GNN", alias="prod"))
+```
+
+- Registration is independent of `export`. If it fails, `fit` warns and returns the result with `registered_model=None`; an invalid alias or a connector that cannot publish models raises before training.
+- A plain `fit(export_config=ModelOutputConfig(connector, location=...))` or `Trainer.export(...)` publishes as before, with or without a block, and also accepts an `alias`; its version is `result.published_model`.
+- An alias is a letter or underscore followed by letters, digits or underscores; `DEFAULT`, `FIRST`, `LAST`, `LATEST` and `v<digits>` are reserved. The alias is set after publishing; if that fails, the publish still stands, with a warning. Snowflake allows one alias per version; if moving an alias fails, it is put back on the version that held it. Snowflake's `DEFAULT`, `FIRST` and `LAST` can be loaded but not set. File, S3 and GCS connectors have no aliases.
+- Each registered version records only pointers: the `kumo_engine` marker and `kumo_bundle_id` (Snowflake version metrics, Databricks version tags). The bundle's own `_provenance.json` holds its lineage. `list_models` and `fetch_model` return `kumo_bundle_id` in `PublishedModel.metadata`; `load_from_connector` warns if the fetched bundle differs from it.
+
 ### Moving snapshots between machines
 
 To build an index on one machine and train or score on another, export it to a file destination and load it there:
diff --git a/packages/kumo-engine/kumo/engine/__init__.py b/packages/kumo-engine/kumo/engine/__init__.py
index b9e467763..3c41f667f 100644
--- a/packages/kumo-engine/kumo/engine/__init__.py
+++ b/packages/kumo-engine/kumo/engine/__init__.py
@@ -62,6 +62,7 @@
     Trainer,
     TrainingResult,
     TrainingTableOutputConfig,
+    set_model_alias,
 )
 
 __all__ = [
@@ -107,6 +108,7 @@
     "PredictionOutputConfig",
     "TrainingTableOutputConfig",
     "SnapshotOutputConfig",
+    "set_model_alias",
 ]
 
 ensure_nofile()
diff --git a/packages/kumo-engine/kumo/engine/trainer/__init__.py b/packages/kumo-engine/kumo/engine/trainer/__init__.py
index abc11c7e3..cddac4046 100644
--- a/packages/kumo-engine/kumo/engine/trainer/__init__.py
+++ b/packages/kumo-engine/kumo/engine/trainer/__init__.py
@@ -13,6 +13,7 @@
     PredictionResult,
     Trainer,
     TrainingResult,
+    set_model_alias,
 )
 
 __all__ = [
@@ -24,4 +25,5 @@
     "PredictionOutputConfig",
     "SnapshotOutputConfig",
     "TrainingTableOutputConfig",
+    "set_model_alias",
 ]
diff --git a/packages/kumo-engine/kumo/engine/trainer/trainer.py b/packages/kumo-engine/kumo/engine/trainer/trainer.py
index fc35048c0..992fd91d0 100644
--- a/packages/kumo-engine/kumo/engine/trainer/trainer.py
+++ b/packages/kumo-engine/kumo/engine/trainer/trainer.py
@@ -44,10 +44,13 @@
 from kumo_core.types import ExplainMethod, PredictionArtifactType
 from kumoapi.model_plan import ModelPlan, TrainingTableGenerationPlan
 from kumoapi.plugin import (
+    BUNDLE_ID_KEY,
+    ModelAliasingSink,
     ModelFetchingSink,
     ModelRegistryLocation,
     PublishedModel,
     SourceConnector,
+    check_model_alias,
 )
 from kumoapi.train import TrainingStage
 from typing_extensions import Self
@@ -86,6 +89,7 @@ class TrainingResult:
 
     `published_model` records the destination's assigned name, version and URI
     when `fit()` exports the bundle. It is `None` without an export.
+    `registered_model` is the version a `kumo.MLOps` ``registry`` registered.
 
     `holdout_path` is the published model's holdout Parquet file written by
     `fit()`, or `None` when it has none (no holdout task, a model refit on the
@@ -102,6 +106,7 @@ class TrainingResult:
     holdout_path: Optional[Path] = None
     holdout_truncated: Optional[bool] = False
     bundle_id: Optional[str] = None
+    registered_model: Optional[PublishedModel] = None
     scoring_defaults: ScoringDefaults = field(default_factory=ScoringDefaults)
     _trainer: Optional['Trainer'] = field(default=None, repr=False,
                                           compare=False)
@@ -331,22 +336,46 @@ def _require_publisher(export_config: ModelOutputConfig) -> None:
         raise NotImplementedError(
             f'{type(export_config.connector).__name__} cannot publish '
             f'models')
+    if export_config.location and export_config.location.alias:
+        check_model_alias(export_config.location.alias)
+
+
+def _connector_registry(settings: Any) -> Optional[ModelOutputConfig]:
+    registry = None if settings is None else settings.registry
+    return registry if isinstance(registry, ModelOutputConfig) else None
 
 
 def _publish(training_result: TrainingResult,
              export_config: ModelOutputConfig) -> PublishedModel:
     spec = read_training_spec(training_result.bundle_dir)
-    return export_config.connector.sink_backend.publish_model(
+    metadata = {
+        name: str(getattr(spec, name))
+        for name in ('spec_version', 'predictive_query', 'snapshot_id')
+    }
+    if training_result.bundle_id:
+        metadata[BUNDLE_ID_KEY] = training_result.bundle_id
+    location = export_config.location
+    alias = location.alias if location else None
+    if location and alias:
+        location = replace(location, alias=None)
+    published = export_config.connector.sink_backend.publish_model(
         export_config.connector,
         str(training_result.bundle_dir),
         path=export_config.path,
-        location=export_config.location,
-        metadata={
-            name: str(getattr(spec, name))
-            for name in ('spec_version', 'predictive_query', 'snapshot_id')
-        },
+        location=location,
+        metadata=metadata,
         metrics=training_result.metrics,
     )
+    if location and alias:
+        try:
+            set_model_alias(export_config.connector,
+                            replace(location, version=published.version),
+                            alias)
+        except Exception as error:
+            warnings.warn(
+                f"Published {published.name} version {published.version} "
+                f"without alias {alias}: {error}", UserWarning, stacklevel=4)
+    return published
 
 
 def _stage_model_bundle(artifacts: Mapping[str, str], dest: Path) -> Path:
@@ -514,6 +543,9 @@ def fit(
         from kumo.engine._mlops import resolve
 
         settings = resolve(mlops)
+        registry = _connector_registry(settings)
+        if registry is not None:
+            _require_publisher(registry)
         run_id = f'{self._id}-fit-{new_run_id()}'
         session = TrackingSession(
             None if settings is None else settings.tracking)
@@ -546,6 +578,20 @@ def fit(
             **result.metrics, 'seconds': time.monotonic() - fit_start
         }, params={'status': 'succeeded'})
         session.finish(result, None)
+        if registry is not None:
+            result = self._register(result, registry)
+        return result
+
+    def _register(self, result: TrainingResult,
+                  registry: ModelOutputConfig) -> TrainingResult:
+        try:
+            registered = _publish(result, registry)
+        except Exception as error:
+            warnings.warn(f"Could not register the model: {error}",
+                          UserWarning, stacklevel=3)
+            return result
+        result = replace(result, registered_model=registered)
+        self._training_result = result
         return result
 
     def _fit(
@@ -1140,10 +1186,24 @@ def load_from_connector(
         if not isinstance(sink, ModelFetchingSink):
             raise TypeError(
                 'model_registry.sink_backend must implement ModelFetchingSink')
-        sink.fetch_model(cast(SourceConnector, model_registry), str(bundle_dir),
-                         path=address.path, location=address.location)
-        return cls._load_bundle(bundle_dir, embedding_models,
-                                accelerator=accelerator)
+        alias = None if location is None else location.alias
+        if alias is not None and not isinstance(sink, ModelAliasingSink):
+            raise NotImplementedError(
+                f'{type(model_registry).__name__} cannot fetch models by '
+                f'alias')
+        published = sink.fetch_model(cast(SourceConnector, model_registry),
+                                     str(bundle_dir), path=address.path,
+                                     location=address.location)
+        trainer = cls._load_bundle(bundle_dir, embedding_models,
+                                   accelerator=accelerator)
+        recorded = (getattr(published, 'metadata', None)
+                    or {}).get(BUNDLE_ID_KEY)
+        if recorded and trainer.bundle_id and recorded != trainer.bundle_id:
+            warnings.warn(
+                f"The registry records bundle {recorded} for this version, "
+                f"but the fetched bundle is {trainer.bundle_id}.", UserWarning,
+                stacklevel=2)
+        return trainer
 
     @classmethod
     def _load_bundle(
@@ -1172,3 +1232,13 @@ def _load_bundle(
                                                       bundle_dir, paths),
                                                   _trainer=trainer)
         return trainer
+
+
+def set_model_alias(connector: Connector, location: ModelRegistryLocation,
+                    alias: str) -> None:
+    """Point a registry alias at ``location``'s exact version."""
+    sink = connector.sink_backend
+    if not isinstance(sink, ModelAliasingSink):
+        raise NotImplementedError(
+            f'{type(connector).__name__} cannot set model aliases')
+    sink.set_model_alias(cast(SourceConnector, connector), location, alias)
diff --git a/packages/kumo-engine/test/test_scoring_defaults.py b/packages/kumo-engine/test/test_scoring_defaults.py
index 497204d17..443b927f2 100644
--- a/packages/kumo-engine/test/test_scoring_defaults.py
+++ b/packages/kumo-engine/test/test_scoring_defaults.py
@@ -4,12 +4,14 @@
 
 import hashlib
 import shutil
+from types import SimpleNamespace
 
 import numpy as np
 import pandas as pd
 import pytest
 from kumo_core.artifacts import add_bundle_file, remove_bundle_file
 from kumo_core.types import TaskType
+from kumoapi.plugin import PublishedModel
 
 import kumo.engine as kumo
 from kumo.engine.trainer._scoring import normalized_scoring_defaults
@@ -99,6 +101,18 @@ def test_defaults_survive_export_and_connector_load(fitted, tmp_path):
     assert loaded.bundle_id == child.id
 
 
+def test_derived_export_publishes_the_child_bundle_id(fitted):
+    trainer, *_ = fitted
+    child = trainer.result.with_scoring_defaults(
+        kumo.ScoringDefaults(binary_classification_threshold=.4))
+    calls = []
+    sink = SimpleNamespace(publish_model=lambda *a, **k: calls.append(k) or
+                           PublishedModel(name='m', version='1', uri='u'))
+    child._trainer.export(
+        kumo.ModelOutputConfig(SimpleNamespace(sink_backend=sink), path='m'))
+    assert calls[0]['metadata']['kumo_bundle_id'] == child.id
+
+
 def test_defaults_are_validated_against_the_task(fitted):
     trainer, *_ = fitted
     with pytest.raises(ValueError, match='num_classes_to_return'):
diff --git a/packages/kumo-engine/test/test_trainer_registry.py b/packages/kumo-engine/test/test_trainer_registry.py
index e668544b2..77d015c25 100644
--- a/packages/kumo-engine/test/test_trainer_registry.py
+++ b/packages/kumo-engine/test/test_trainer_registry.py
@@ -238,7 +238,11 @@ def test_finish_fit_forwards_publication_details_and_returns_assigned_version(
     class Sink:
         def publish_model(self, connector, bundle_dir, **kwargs):
             spec = json.loads((Path(bundle_dir) / SPEC_FILE).read_text())
-            assert kwargs['metadata'] == {
+            assert {
+                key: value
+                for key, value in kwargs['metadata'].items()
+                if not key.startswith('kumo_')
+            } == {
                 name: str(spec[name])
                 for name in ('spec_version', 'predictive_query', 'snapshot_id')
             }
@@ -258,6 +262,7 @@ def publish_model(self, connector, bundle_dir, **kwargs):
     assert calls[0][2]['path'] == path
     assert calls[0][2]['location'] == location
     assert calls[0][2]['metrics'] == {'val_auroc': .9}
+    assert calls[0][2]['metadata']['kumo_bundle_id'] == result.bundle_id
     assert result.published_model is published
     assert result.metrics == original.metrics
     assert result.bundle_dir == bundle
@@ -572,3 +577,165 @@ def fake_predict(**kwargs) -> None:
                              binary_classification_threshold=.5)
 
     assert result.frame['TARGET_PRED'].tolist() == [True]
+
+
+def _export(sink, bundle, location=None, artifact_paths=None):
+    return Trainer(ModelPlan())._finish_fit(
+        TrainingResult({}, artifact_paths or {},
+                       bundle), predictive_query='PREDICT COUNT(*)',
+        snapshot_id='snap-123', training_table_generation=_generation_plan(),
+        export_config=SimpleNamespace(
+            connector=SimpleNamespace(sink_backend=sink),
+            path=None if location else 'm', location=location))
+
+
+def _recording_sink(calls):
+    return SimpleNamespace(publish_model=lambda *a, **k: calls.append(k) or
+                           PublishedModel(name='m', version='1', uri='u'))
+
+
+def test_publish_records_only_the_bundle_id_pointer(tmp_path: Path) -> None:
+    calls = []
+    result = _export(_recording_sink(calls), _fake_published_bundle(tmp_path))
+    assert {k: v
+            for k, v in calls[0]['metadata'].items()
+            if k.startswith('kumo_')} == {'kumo_bundle_id': result.bundle_id}
+
+
+class _AliasingSink(_FakeSink):
+    def __init__(self, recorded: Optional[str] = None,
+                 fail: bool = False) -> None:
+        super().__init__()
+        self.recorded = recorded
+        self.fail = fail
+        self.aliases: dict = {}
+
+    def set_model_alias(self, connector, location, alias) -> None:
+        if self.fail:
+            raise RuntimeError('registry unavailable')
+        self.aliases[alias] = location.version
+
+    def fetch_model(self, connector, local_dir, **kwargs) -> PublishedModel:
+        shutil.copytree(self.published['CHURN_GNN@latest'], local_dir,
+                        dirs_exist_ok=True)
+        return PublishedModel(name='CHURN_GNN', version='V1', uri='fake://',
+                              metadata={'kumo_bundle_id': self.recorded})
+
+
+@pytest.mark.parametrize('sink,warning', [
+    (_AliasingSink(), None),
+    (_AliasingSink(fail=True), 'registry unavailable'),
+    (_FakeSink(), 'cannot set model aliases'),
+])
+def test_export_with_alias_publishes_then_sets_it_best_effort(
+        tmp_path: Path, sink, warning) -> None:
+    import warnings
+
+    with warnings.catch_warnings(record=True) as caught:
+        warnings.simplefilter('always')
+        result = _export(sink, _fake_published_bundle(tmp_path),
+                         ModelRegistryLocation('CHURN_GNN', alias='prod'))
+    assert result.published_model.version == 'V1'
+    assert list(sink.published) == ['CHURN_GNN@latest']
+    messages = [str(w.message) for w in caught if 'alias' in str(w.message)]
+    if warning is None:
+        assert sink.aliases == {'prod': 'V1'} and messages == []
+    else:
+        assert len(messages) == 1 and warning in messages[0]
+
+
+def test_load_by_alias_warns_on_recorded_bundle_mismatch(
+        tmp_path: Path) -> None:
+    bundle = _fake_published_bundle(tmp_path)
+    write_training_spec(bundle, predictive_query='PREDICT COUNT(*)',
+                        model_plan=ModelPlan(), snapshot_id='snap-123',
+                        training_table_generation=_generation_plan())
+    sink = _AliasingSink('other')
+    sink.publish_model(None, str(bundle),
+                       location=ModelRegistryLocation('CHURN_GNN'))
+    with pytest.warns(UserWarning, match='registry records bundle other'):
+        loaded = Trainer.load_from_connector(
+            _FakeConnector(sink),
+            location=ModelRegistryLocation('CHURN_GNN', alias='prod'))
+    assert loaded.bundle_id == Trainer.load(bundle).bundle_id
+
+
+def test_alias_needs_an_aliasing_sink() -> None:
+    from kumo.engine import set_model_alias
+
+    connector = _FakeConnector(_FakeSink())
+    with pytest.raises(NotImplementedError, match='alias'):
+        Trainer.load_from_connector(
+            connector, location=ModelRegistryLocation('m', alias='prod'))
+    with pytest.raises(NotImplementedError, match='alias'):
+        set_model_alias(connector, ModelRegistryLocation('m', version='1'),
+                        'prod')
+    for kwargs in (dict(alias='prod', version='3'), dict(alias='')):
+        with pytest.raises(ValueError, match='alias'):
+            ModelRegistryLocation('m', **kwargs)
+
+
+def test_invalid_export_alias_is_rejected_before_publishing(
+        tmp_path: Path) -> None:
+    sink = _AliasingSink()
+    trainer = Trainer(ModelPlan())
+    trainer._training_result = TrainingResult({}, {},
+                                              _fake_published_bundle(tmp_path))
+    with pytest.raises(ValueError, match='model alias'):
+        trainer.export(
+            SimpleNamespace(
+                connector=SimpleNamespace(sink_backend=sink), path=None,
+                location=ModelRegistryLocation('CHURN_GNN', alias='LATEST')))
+    assert sink.published == {}
+
+
+@pytest.mark.parametrize('sink,registered', [
+    (_AliasingSink(), 'V1'),
+    (SimpleNamespace(publish_model=lambda *a, **k: 1 / 0), None),
+])
+def test_mlops_registry_registers_after_the_fit(tmp_path: Path, monkeypatch,
+                                                sink, registered) -> None:
+    import warnings
+
+    import kumo.engine as kumo
+
+    bundle = _fake_published_bundle(tmp_path)
+    write_training_spec(bundle, predictive_query='PREDICT COUNT(*)',
+                        model_plan=ModelPlan(), snapshot_id='snap-123',
+                        training_table_generation=_generation_plan())
+    monkeypatch.setattr(
+        Trainer, '_fit', lambda self, *a, **k: TrainingResult(
+            {}, {}, bundle, bundle_id='b1'))
+    registry = kumo.ModelOutputConfig(
+        _FakeConnector(sink),
+        location=ModelRegistryLocation('CHURN_GNN', alias='prod'))
+    trainer = Trainer(ModelPlan())
+    with warnings.catch_warnings(record=True) as caught:
+        warnings.simplefilter('always')
+        result = trainer.fit(
+            SimpleNamespace(query='PREDICT COUNT(*)'),
+            snapshot=SimpleNamespace(id='s'),
+            mlops=kumo.MLOps(str(tmp_path / 'root'), registry=registry,
+                             export=False))
+    failures = [w for w in caught if 'Could not register' in str(w.message)]
+    assert getattr(result.registered_model, 'version', None) == registered
+    assert result.published_model is None
+    if registered:
+        assert trainer.result is result and not failures
+        assert sink.aliases == {'prod': 'V1'}
+    else:
+        assert len(failures) == 1
+
+
+def test_mlops_registry_alias_is_checked_before_the_fit(tmp_path: Path,
+                                                        monkeypatch) -> None:
+    import kumo.engine as kumo
+
+    monkeypatch.setattr(Trainer, '_fit', lambda *a, **k: 1 / 0)
+    registry = kumo.ModelOutputConfig(
+        _FakeConnector(_AliasingSink()),
+        location=ModelRegistryLocation('CHURN_GNN', alias='LATEST'))
+    with pytest.raises(ValueError, match='model alias'):
+        Trainer(ModelPlan()).fit(
+            SimpleNamespace(query='q'), snapshot=SimpleNamespace(id='s'),
+            mlops=kumo.MLOps(str(tmp_path), registry=registry, export=False))
diff --git a/packages/kumo-snowflake/src/kumo_snowflake/registry.py b/packages/kumo-snowflake/src/kumo_snowflake/registry.py
index fd220653e..16d14d983 100644
--- a/packages/kumo-snowflake/src/kumo_snowflake/registry.py
+++ b/packages/kumo-snowflake/src/kumo_snowflake/registry.py
@@ -33,11 +33,13 @@
 """
 
 import json
+import logging
 import shutil
 from contextlib import contextmanager
 from pathlib import Path
 from typing import (
     TYPE_CHECKING,
+    Any,
     ContextManager,
     Dict,
     Iterator,
@@ -53,9 +55,13 @@
     resolved,
 )
 from kumoapi.plugin import (
+    BUNDLE_ID_KEY,
+    ENGINE_MARKER,
     ModelRegistryLocation,
     PublishedModel,
     SourceConnector,
+    check_model_alias,
+    registry_lineage,
 )
 
 if TYPE_CHECKING:
@@ -74,8 +80,9 @@
 #: merely happens to be named alike.
 _MANIFEST_FILE = 'publication-manifest.json'
 
-#: Engine-bundle marker stored through the registry metadata API.
-_ENGINE_MARKER = 'kumo_engine'
+_SYSTEM_ALIASES = frozenset({'DEFAULT', 'FIRST', 'LAST'})
+
+_LOG = logging.getLogger(__name__)
 
 #: Where a fetched version is unpacked before its bundle is lifted out. Named
 #: rather than a temporary directory so a half-finished import leaves something
@@ -209,6 +216,40 @@ def _refuse_path(path: Optional[str], action: str) -> None:
             f'a path.')
 
 
+def _version_record(metadata: Mapping[str, str]) -> Dict[str, str]:
+    record = {ENGINE_MARKER: '1'}
+    if metadata.get(BUNDLE_ID_KEY):
+        record[BUNDLE_ID_KEY] = str(metadata[BUNDLE_ID_KEY])
+    return record
+
+
+def _recorded_lineage(version: "ModelVersion") -> Dict[str, str]:
+    try:
+        return registry_lineage(dict(version.show_metrics() or {}))
+    except Exception as error:
+        _LOG.warning('Could not read registry metadata: %s', error)
+        return {}
+
+
+def _alias_holders(model: Any) -> Dict[str, str]:
+    holders = {}
+    for row in model.show_versions().to_dict('records'):
+        for alias in json.loads(row.get('aliases') or '[]'):
+            holders[alias] = str(row['name'])
+    return holders
+
+
+def _restore_alias(model: Any, version: str, alias: str) -> None:
+    try:
+        model.version(version).set_alias(alias)
+    except Exception as error:
+        _LOG.warning('Alias %s is now unset: restoring it on version %s '
+                     'failed: %s', alias, version, error)
+    else:
+        _LOG.warning('Could not move alias %s; it stays on version %s.',
+                     alias, version)
+
+
 class SnowflakeModelRegistryMixin:
     r"""The model half of :class:`SnowflakeSinkBackend`.
 
@@ -267,8 +308,8 @@ def publish_model(
                 the model, and optionally the version. A version of ``None``
                 lets the registry assign one -- the usual case, and the reason
                 this reports back what it created.
-            metadata: Returned with the publication result. Training metadata
-                remains in the bundle and is not recorded in registry tags.
+            metadata: Returned with the publication result; its
+                ``kumo_bundle_id`` is also recorded in the version's metrics.
             metrics: Returned with the publication result. Training metrics
                 remain in the bundle's metrics.json.
             project_id: Ignored; a local runtime has no projects.
@@ -288,13 +329,14 @@ def publish_model(
                 f"{bundle} does not look like a published model bundle: it "
                 f"has no {_SPEC_FILE}, which an export writes into every one.")
 
+        record = _version_record(metadata)
         with self._registry(connector) as registry:
             version = registry.log_model(
                 _bundle_model(str(bundle)),
                 model_name=location.model_name,
                 version_name=location.version,
                 comment=location.comment,
-                metrics={_ENGINE_MARKER: '1'},
+                metrics=record,
                 signatures=_carrier_signature(),
             )
         return _published(connector, version, metadata=metadata,
@@ -308,10 +350,11 @@ def list_models(
     ) -> List[PublishedModel]:
         r"""Lists Kumo model versions visible in the connector's registry.
 
-        A Kumo export records only ``kumo_engine=1`` in registry metadata.
+        A Kumo export records ``kumo_engine=1`` in registry metadata.
         That marker makes discovery both cheap and precise:
         versions logged by another tool are omitted, and no bundle is
         downloaded merely to decide whether it belongs in this list.
+        ``metadata`` carries the recorded ``kumo_bundle_id``.
         """
         del project_id
         found: List[PublishedModel] = []
@@ -319,9 +362,11 @@ def list_models(
             for model in registry.models():
                 for version in model.versions():
                     recorded = dict(version.show_metrics() or {})
-                    if recorded.get(_ENGINE_MARKER) != '1':
+                    if recorded.get(ENGINE_MARKER) != '1':
                         continue
-                    found.append(_published(connector, version))
+                    found.append(
+                        _published(connector, version,
+                                   metadata=registry_lineage(recorded)))
         return sorted(found, key=lambda item: (item.name, item.version))
 
     def fetch_model(
@@ -345,14 +390,15 @@ def fetch_model(
                 though the caller always hands this a fresh, empty
                 ``mkdtemp``.
             path: Refused, as in :meth:`publish_model`.
-            location: The model, and optionally the version. ``None`` takes the
-                registry's default version, which is what a caller who names
-                none means.
+            location: The model, and optionally the version or an alias. With
+                neither, takes the registry's default version, which is what a
+                caller who names none means.
             project_id: Ignored.
 
         Returns:
             The name, the version actually read -- resolved, when the caller
-            named none -- where it lives, and metrics read from metrics.json.
+            named none -- where it lives, metrics read from metrics.json, and
+            the version's recorded ``kumo_bundle_id`` as ``metadata``.
         """
         del project_id
         _refuse_path(path, 'fetching')
@@ -366,8 +412,18 @@ def fetch_model(
             # Resolved before anything on disk is touched: a name the registry
             # does not know must not cost the caller the bundle already there.
             model = registry.get_model(location.model_name)
-            version = (model.default if location.version is None else
-                       model.version(location.version))
+            alias = getattr(location, 'alias', None)
+            if alias is not None:
+                if resolved(alias) not in _alias_holders(model):
+                    raise ValueError(
+                        f'No version of {location.model_name} holds alias '
+                        f'{alias}.')
+                version = model.version(alias)
+            elif location.version is None:
+                version = model.default
+            else:
+                version = model.version(location.version)
+            recorded = _recorded_lineage(version)
 
             destination.parent.mkdir(parents=True, exist_ok=True)
             staged = destination.parent / f'{destination.name}{_EXPORT_DIR}'
@@ -385,9 +441,52 @@ def fetch_model(
                 isinstance(key, str) and isinstance(value, (int, float))
                 for key, value in metrics.items())):
             raise ValueError('Invalid training metrics in model bundle.')
-        published = _published(connector, version, metrics=metrics)
+        published = _published(connector, version, metadata=recorded,
+                               metrics=metrics)
         if destination.exists():
             shutil.rmtree(destination)
         shutil.move(str(bundle), str(destination))
         shutil.rmtree(staged, ignore_errors=True)
         return published
+
+    def set_model_alias(
+        self,
+        connector: SourceConnector,
+        location: ModelRegistryLocation,
+        alias: str,
+    ) -> None:
+        r"""Points :obj:`alias` at ``location.version``, moving it if held.
+
+        Snowflake allows one alias per version. A failed move puts the alias
+        back on the version that held it.
+        """
+        if location.version is None:
+            raise ValueError('Setting an alias needs a ModelRegistryLocation '
+                             'naming an exact version.')
+        check_model_alias(alias)
+        wanted = resolved(alias)
+        with self._registry(connector) as registry:
+            model = registry.get_model(location.model_name)
+            target = model.version(location.version)
+            name = resolved(str(target.version_name))
+            holders = {
+                held: version
+                for held, version in _alias_holders(model).items()
+                if held not in _SYSTEM_ALIASES
+            }
+            if holders.get(wanted) == name:
+                return
+            held = sorted(a for a, v in holders.items() if v == name)
+            if held:
+                raise ValueError(
+                    f'Version {name} of {location.model_name} already holds '
+                    f'alias {held[0]}; unset it first.')
+            previous = holders.get(wanted)
+            if previous is not None:
+                model.version(previous).unset_alias(alias)
+            try:
+                target.set_alias(alias)
+            except Exception:
+                if previous is not None:
+                    _restore_alias(model, previous, alias)
+                raise
diff --git a/packages/kumo-snowflake/tests/test_registry.py b/packages/kumo-snowflake/tests/test_registry.py
index 9d07e6c28..9ed7f1c94 100644
--- a/packages/kumo-snowflake/tests/test_registry.py
+++ b/packages/kumo-snowflake/tests/test_registry.py
@@ -8,6 +8,7 @@
 from pathlib import Path
 from types import SimpleNamespace
 
+import pandas as pd
 import pytest
 from kumo_snowflake import registry as registry_module
 from kumo_snowflake.registry import SnowflakeModelRegistryMixin
@@ -30,7 +31,11 @@ def export(directory):
     version = SimpleNamespace(
         model_name='CHURN', version_name='V1',
         fully_qualified_model_name='DB.SCHEMA.CHURN', export=export,
-        show_metrics=lambda: pytest.fail('read registry metrics'))
+        show_metrics=lambda: {
+            'kumo_engine': '1',
+            'kumo_bundle_id': 'b1',
+            'auc': 0.5
+        })
     model = SimpleNamespace(default=version, version=lambda _: version)
 
     @contextmanager
@@ -60,6 +65,7 @@ def test_fetch_returns_bundle_metrics(fetch, requested_version, metrics):
     payload = json.dumps(metrics) if metrics is not None else None
     published = run(payload, requested_version)
     assert published.metrics == (metrics or {})
+    assert published.metadata == {'kumo_bundle_id': 'b1'}
     assert published.version == 'V1'
     assert (destination / 'weights.bin').read_bytes() == b'model weights'
     assert not (destination / 'previous').exists()
@@ -75,9 +81,13 @@ def test_invalid_metrics_preserve_previous_destination(fetch, payload):
     assert (destination / 'previous').read_bytes() == b'previous bundle'
 
 
-@pytest.mark.parametrize('metadata', [{}, {'predictive_query': 'x' * 300}])
-def test_publish_records_only_engine_marker_in_catalog(tmp_path, monkeypatch,
-                                                       metadata):
+@pytest.mark.parametrize('metadata', [{}, {
+    'predictive_query': 'x' * 300,
+    'kumo_bundle_id': 'b1',
+    'kumo_engine': '0'
+}])
+def test_publish_records_only_pointers_in_catalog(
+        tmp_path, monkeypatch, metadata):
     bundle = tmp_path / 'bundle'
     bundle.mkdir()
     (bundle / 'training-spec.json').write_text('{}')
@@ -99,16 +109,20 @@ def registry(_):
     monkeypatch.setattr(registry_module, '_carrier_signature', lambda: {})
     published = sink.publish_model(SimpleNamespace(), str(bundle),
                                    location=ModelRegistryLocation('CHURN'),
-                                   metadata=metadata, metrics={'auc': 0.92})
-    assert calls[0]['metrics'] == {'kumo_engine': '1'}
+                                   metadata=metadata,
+                                   metrics={'auc': 0.92, 'loss': float('nan')})
+    assert calls[0]['metrics'] == {
+        **({'kumo_bundle_id': 'b1'} if metadata else {}), 'kumo_engine': '1'
+    }
     assert published.metadata == metadata
-    assert published.metrics == {'auc': 0.92}
+    assert published.metrics['auc'] == 0.92
 
 
-def test_listing_filters_engine_marker_and_returns_empty_metrics(monkeypatch):
+def test_listing_filters_engine_marker_and_returns_lineage(monkeypatch):
     versions = []
     for name, recorded in [('V1', {
             'kumo_engine': '1',
+            'kumo_bundle_id': 'b1',
             'auc': 0.92
     }), ('V2', {
             'other': '1',
@@ -134,4 +148,112 @@ def registry(_):
     found = sink.list_models(SimpleNamespace())
     assert [item.version for item in found] == ['V1', 'V4']
     assert [item.metrics for item in found] == [{}, {}]
-    assert all(item.metadata == {} for item in found)
+    assert [item.metadata for item in found] == [{'kumo_bundle_id': 'b1'}, {}]
+
+
+class _Model:
+    def __init__(self, aliases):
+        self.aliases = aliases
+
+    def show_versions(self):
+        return pd.DataFrame([
+            dict(name=name, aliases=json.dumps(held))
+            for name, held in self.aliases.items()
+        ])
+
+    def version(self, version_or_alias):
+        name = next((name for name, held in self.aliases.items()
+                     if version_or_alias.upper() in held), version_or_alias)
+        held = self.aliases[name]
+
+        def export(directory):
+            bundle = Path(directory) / 'bundle'
+            bundle.mkdir()
+            (bundle / 'training-spec.json').write_text(name)
+            (bundle / 'publication-manifest.json').write_text('{}')
+
+        return SimpleNamespace(
+            version_name=name, model_name='CHURN',
+            fully_qualified_model_name='DB.SCHEMA.CHURN', export=export,
+            set_alias=lambda alias: held.append(alias.upper()),
+            unset_alias=lambda alias: held.remove(alias.upper()),
+            show_metrics=lambda: {
+                'kumo_engine': '1',
+                'kumo_bundle_id': f'id-{name}'
+            })
+
+
+def _aliasing_sink(monkeypatch, model):
+    @contextmanager
+    def registry(_):
+        yield SimpleNamespace(get_model=lambda _: model)
+
+    sink = SnowflakeModelRegistryMixin()
+    monkeypatch.setattr(sink, '_registry', registry)
+    return sink
+
+
+def test_set_alias_moves_it_and_refuses_what_snowflake_cannot(monkeypatch):
+    from kumo_snowflake.sink import SnowflakeSinkBackend
+    from kumoapi.plugin import ModelAliasingSink
+
+    assert isinstance(SnowflakeSinkBackend(), ModelAliasingSink)
+    model = _Model({'V1': ['FIRST'], 'V2': ['LAST', 'DEFAULT'], 'V3': []})
+    sink = _aliasing_sink(monkeypatch, model)
+
+    def set_alias(version, alias):
+        sink.set_model_alias(SimpleNamespace(),
+                             ModelRegistryLocation('CHURN', version=version),
+                             alias)
+
+    set_alias('V2', 'prod')
+    set_alias('V3', 'prod')
+    set_alias('V3', 'prod')
+    assert model.aliases == {
+        'V1': ['FIRST'],
+        'V2': ['LAST', 'DEFAULT'],
+        'V3': ['PROD']
+    }
+    for version, alias, match in [('V3', 'staging', 'already holds'),
+                                  ('V1', 'default', 'reserved'),
+                                  (None, 'prod', 'exact version')]:
+        with pytest.raises(ValueError, match=match):
+            set_alias(version, alias)
+    assert model.aliases['V3'] == ['PROD']
+
+
+def test_fetch_by_alias_reads_the_aliased_version(monkeypatch, tmp_path):
+    sink = _aliasing_sink(monkeypatch, _Model({
+        'V1': ['DEFAULT'],
+        'V2': ['PROD']
+    }))
+    published = sink.fetch_model(
+        SimpleNamespace(), str(tmp_path / 'out'),
+        location=ModelRegistryLocation('CHURN', alias='prod'))
+    assert published.version == 'V2'
+    assert published.metadata == {'kumo_bundle_id': 'id-V2'}
+    assert (tmp_path / 'out' / 'training-spec.json').read_text() == 'V2'
+    with pytest.raises(ValueError, match='holds alias'):
+        sink.fetch_model(SimpleNamespace(), str(tmp_path / 'other'),
+                         location=ModelRegistryLocation('CHURN', alias='V1'))
+
+
+def test_failed_alias_move_puts_it_back(monkeypatch):
+    model = _Model({'V1': ['PROD'], 'V2': []})
+
+    def fail(alias):
+        raise RuntimeError('lost connection')
+
+    def failing_version(name):
+        version = _Model.version(model, name)
+        if name == 'V2':
+            version.set_alias = fail
+        return version
+
+    model.version = failing_version
+    sink = _aliasing_sink(monkeypatch, model)
+    with pytest.raises(RuntimeError, match='lost connection'):
+        sink.set_model_alias(SimpleNamespace(),
+                             ModelRegistryLocation('CHURN', version='V2'),
+                             'prod')
+    assert model.aliases == {'V1': ['PROD'], 'V2': []}