fix(BA-6015): carry mount metadata in RevisionDraft merge chain

Mount identity (model vfolder, definition path, destination, extra
mounts, subpath) now travels inside RevisionDraft and participates in
the merge chain instead of being threaded as a sidecar argument through
add_revision and the draft readers. The vfolder scan records the
matched candidate path back into the merged draft, so the resolved
model-definition path is persisted on the revision instead of the
request's possibly-empty value. ModelRevisionCreator.to_draft is
renamed to to_draft_with_extra_mount and now derives the rest of the
mount metadata itself, taking only the permission-resolved extra mounts.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

Author: seedspirit

src/ai/backend/manager/data/deployment/creator.py

@@ -8,6 +8,7 @@
 from ai.backend.common.identifier.deployment_preset import DeploymentPresetID
 from ai.backend.common.identifier.image import ImageID
 from ai.backend.common.identifier.vfolder import VFolderUUID
+from ai.backend.common.types import MountInfoEntry
 from ai.backend.manager.data.deployment.types import (
     DeploymentMetadata,
     DeploymentNetworkSpec,
@@ -17,6 +18,7 @@
     ModelRevisionSpec,
     ModelRevisionSpecDraft,
     MountInfo,
+    MountMetadata,
     ReplicaSpec,
     ResourceSpec,
     RevisionDraft,
@@ -65,7 +67,7 @@ class ModelRevisionCreator:
     revision_preset_id: DeploymentPresetID | None = None
     preset_values: list[PresetValueData] = field(default_factory=list)
 
-    def to_draft(self) -> RevisionDraft:
+    def to_draft_with_extra_mount(self, extra_mounts: list[MountInfoEntry]) -> RevisionDraft:
         """Project this v2 creator onto a ``RevisionDraft`` layer.
 
         ``image_id`` is already resolved upstream. Optional ``resource_spec`` /
@@ -77,6 +79,13 @@ def to_draft(self) -> RevisionDraft:
         ex = self.execution
         return RevisionDraft(
             image_id=self.image_id,
+            mounts=MountMetadata(
+                model_vfolder_id=self.mounts.model_vfolder_id,
+                model_definition_path=self.mounts.model_definition_path,
+                model_mount_destination=self.mounts.model_mount_destination,
+                extra_mounts=extra_mounts,
+                vfolder_subpath=self.mounts.vfolder_subpath,
+            ),
             resource_slots=rs.resource_slots if rs is not None else None,
             resource_opts=rs.resource_opts if rs is not None else None,
             cluster_mode=rs.cluster_mode if rs is not None else None,

src/ai/backend/manager/data/deployment/types.py

@@ -67,6 +67,17 @@ class DeploymentConfig(BackendAISchema):
     environ: dict[str, str] | None = None
 
 
+@dataclass(frozen=True)
+class FetchedModelDefinition:
+    """Model definition draft with the vfolder path it was read from.
+
+    ``path`` is the matched candidate path inside the model vfolder.
+    """
+
+    path: str
+    model_definition: ModelDefinitionDraft
+
+
 class RouteStatus(enum.Enum):
     """Lifecycle status of a route (independent of health)."""
 
@@ -512,12 +523,11 @@ def to_draft(self, image_id: ImageID | None) -> RevisionDraft:
         """Project this legacy spec draft onto a ``RevisionDraft`` layer.
 
         ``image_id`` is resolved upstream from the spec's ``image_identifier``
-        (canonical + architecture) via the repository; mount-identifying
-        fields live on ``MountMetadata`` and are passed alongside into
-        ``add_revision`` rather than through the draft chain.
+        (canonical + architecture) via the repository.
         """
         return RevisionDraft(
             image_id=image_id,
+            mounts=self.mounts,
             resource_slots=self.resource_spec.resource_slots,
             resource_opts=self.resource_spec.resource_opts,
             cluster_mode=self.resource_spec.cluster_mode,
@@ -549,6 +559,8 @@ class RevisionDraft:
     # (see ``deployment_revisions.image`` SET NULL FK) — downstream
     # resolvers must treat the latter as non-deployable.
     image_id: ImageID | None = None
+    # Mount
+    mounts: MountMetadata | None = None
     # Resource
     resource_slots: Mapping[str, Any] | None = None
     resource_opts: Mapping[str, Any] | None = None
@@ -571,6 +583,7 @@ def merge(self, other: RevisionDraft) -> RevisionDraft:
         """Return a new draft with ``other`` layered on top of ``self``."""
         return RevisionDraft(
             image_id=other.image_id if other.image_id is not None else self.image_id,
+            mounts=_merge_mounts(self.mounts, other.mounts),
             resource_slots=_merge_mappings(self.resource_slots, other.resource_slots),
             resource_opts=_merge_mappings(self.resource_opts, other.resource_opts),
             cluster_mode=other.cluster_mode
@@ -601,7 +614,7 @@ def merge(self, other: RevisionDraft) -> RevisionDraft:
             else (list(self.preset_values) if self.preset_values is not None else None),
         )
 
-    def to_model_revision_spec(self, mounts: MountMetadata) -> ModelRevisionSpec:
+    def to_model_revision_spec(self) -> ModelRevisionSpec:
         """Project the merged draft into a final ``ModelRevisionSpec``.
 
         Validates that the merge chain produced an ``image_id`` and a
@@ -612,6 +625,8 @@ def to_model_revision_spec(self, mounts: MountMetadata) -> ModelRevisionSpec:
         """
         if self.image_id is None:
             raise InvalidAPIParameters("image_id is required to build a revision")
+        if self.mounts is None:
+            raise InvalidAPIParameters("mounts are required to build a revision")
         if self.runtime_variant_id is None:
             raise InvalidAPIParameters("runtime_variant_id is required to build a revision")
         return ModelRevisionSpec(
@@ -622,7 +637,7 @@ def to_model_revision_spec(self, mounts: MountMetadata) -> ModelRevisionSpec:
                 resource_slots=self.resource_slots or {},
                 resource_opts=self.resource_opts,
             ),
-            mounts=mounts,
+            mounts=self.mounts,
             execution=ExecutionSpec(
                 startup_command=self.startup_command,
                 bootstrap_script=self.bootstrap_script,
@@ -637,6 +652,33 @@ def to_model_revision_spec(self, mounts: MountMetadata) -> ModelRevisionSpec:
         )
 
 
+def _merge_mounts(
+    lower: MountMetadata | None,
+    upper: MountMetadata | None,
+) -> MountMetadata | None:
+    if upper is None:
+        return lower
+    if lower is None:
+        return MountMetadata(
+            model_vfolder_id=upper.model_vfolder_id,
+            model_definition_path=upper.model_definition_path,
+            model_mount_destination=upper.model_mount_destination,
+            extra_mounts=list(upper.extra_mounts),
+            vfolder_subpath=upper.vfolder_subpath,
+        )
+    return MountMetadata(
+        model_vfolder_id=upper.model_vfolder_id,
+        model_definition_path=upper.model_definition_path
+        if upper.model_definition_path is not None
+        else lower.model_definition_path,
+        model_mount_destination=upper.model_mount_destination,
+        extra_mounts=list(upper.extra_mounts),
+        vfolder_subpath=upper.vfolder_subpath
+        if upper.vfolder_subpath is not None
+        else lower.vfolder_subpath,
+    )
+
+
 def _merge_mappings(
     lower: Mapping[str, Any] | None,
     upper: Mapping[str, Any] | None,
@@ -1007,6 +1049,17 @@ def to_draft(self) -> RevisionDraft:
         )
         return RevisionDraft(
             image_id=self.image_id,
+            mounts=(
+                MountMetadata(
+                    model_vfolder_id=self.model_mount_config.vfolder_id,
+                    model_definition_path=self.model_mount_config.definition_path or None,
+                    model_mount_destination=self.model_mount_config.mount_destination or "/models",
+                    extra_mounts=list(self.model_mount_config.extra_mounts),
+                    vfolder_subpath=self.model_mount_config.subpath,
+                )
+                if self.model_mount_config.vfolder_id is not None
+                else None
+            ),
             resource_slots=resource_slots,
             resource_opts=resource_opts,
             cluster_mode=self.cluster_config.mode,

src/ai/backend/manager/repositories/deployment/repository.py

@@ -14,7 +14,7 @@
 from ai.backend.common.clients.valkey_client.valkey_live.client import ValkeyLiveClient
 from ai.backend.common.clients.valkey_client.valkey_schedule.client import ValkeyScheduleClient
 from ai.backend.common.clients.valkey_client.valkey_stat.client import ValkeyStatClient
-from ai.backend.common.config import ModelDefinitionDraft, ModelHealthCheck
+from ai.backend.common.config import ModelHealthCheck
 from ai.backend.common.data.endpoint.types import EndpointLifecycle
 from ai.backend.common.exception import BackendAIError, InvalidAPIParameters
 from ai.backend.common.identifier.deployment import DeploymentID
@@ -64,6 +64,7 @@
     DeploymentRevisionReadBundle,
     DeploymentSummarySearchResult,
     DeploymentWithHistory,
+    FetchedModelDefinition,
     LegacyRevisionCreateReadBundle,
     ModelDeploymentAccessTokenData,
     ModelDeploymentAutoScalingRuleData,
@@ -485,12 +486,13 @@ async def fetch_model_definition(
         self,
         vfolder_id: VFolderUUID,
         model_definition_path: str | None,
-    ) -> ModelDefinitionDraft | None:
+    ) -> FetchedModelDefinition | None:
         """Fetch and validate the model-definition file from the model vfolder.
 
-        Returns a ``ModelDefinitionDraft`` because the file is user-authored
+        Returns a ``FetchedModelDefinition`` because the file is user-authored
         and may omit optional fields; strict-field validation is deferred to
-        the persistence boundary where the merged result is resolved.
+        the persistence boundary where the merged result is resolved. The
+        result also carries the exact candidate path that matched.
         Returns ``None`` when no candidate file exists.
         """
         vfolder_location = await self._db_source.get_vfolder_by_id(vfolder_id)

src/ai/backend/manager/repositories/deployment/storage_source/storage_source.py

@@ -1,6 +1,7 @@
 """Storage source implementation for deployment repository."""
 
 from collections.abc import Mapping
+from dataclasses import dataclass
 from typing import Any, override
 
 import tomli
@@ -9,10 +10,17 @@
 from ai.backend.common.config import ModelDefinitionDraft
 from ai.backend.common.exception import BackendAIError, InvalidAPIParameters
 from ai.backend.common.types import BackendAISchema, SchemaValidationFailureInfo, VFolderID
+from ai.backend.manager.data.deployment.types import FetchedModelDefinition
 from ai.backend.manager.data.vfolder.types import VFolderLocation
 from ai.backend.manager.models.storage import StorageSessionManager
 
 
+@dataclass(frozen=True)
+class FetchedConfigFile:
+    filename: str
+    payload: Mapping[str, Any]
+
+
 class DeploymentConfigInput(BackendAISchema):
     """Validated ``deployment-config.yaml`` / ``service-definition.toml`` payload.
 
@@ -60,13 +68,13 @@ async def fetch_deployment_config(
         raw = await self._fetch_config_file_in_candidates(vfolder_location, candidates)
         if raw is None:
             return None
-        return DeploymentConfigInput.model_validate(dict(raw))
+        return DeploymentConfigInput.model_validate(dict(raw.payload))
 
     async def fetch_model_definition(
         self,
         vfolder_location: VFolderLocation,
         candidates: list[str],
-    ) -> ModelDefinitionDraft | None:
+    ) -> FetchedModelDefinition | None:
         """Fetch the first existing model-definition file among ``candidates``.
 
         Uses the draft type because user-authored files may supply only a
@@ -77,13 +85,16 @@ async def fetch_model_definition(
         raw = await self._fetch_config_file_in_candidates(vfolder_location, candidates)
         if raw is None:
             return None
-        return ModelDefinitionDraft.model_validate(dict(raw))
+        return FetchedModelDefinition(
+            path=raw.filename,
+            model_definition=ModelDefinitionDraft.model_validate(dict(raw.payload)),
+        )
 
     async def _fetch_config_file_in_candidates(
         self,
         vfolder_location: VFolderLocation,
         candidates: list[str],
-    ) -> Mapping[str, Any] | None:
+    ) -> FetchedConfigFile | None:
         """Return the first parsed ``Mapping`` among the candidates.
 
         Candidates are tried in priority order (new name first, legacy
@@ -122,5 +133,5 @@ async def _fetch_config_file_in_candidates(
                 raise InvalidAPIParameters(
                     f"Invalid config file '{filename}': top-level value must be a mapping."
                 )
-            return loaded
+            return FetchedConfigFile(filename=filename, payload=loaded)
         return None

src/ai/backend/manager/services/model_serving/services/model_serving.py

@@ -829,38 +829,30 @@ async def modify_endpoint(self, action: ModifyEndpointAction) -> ModifyEndpointA
         if spec.has_revision_changes():
             latest_rev = await self._deployment_repository.get_latest_revision(action.deployment_id)
             latest_draft = latest_rev.to_draft()
+            if latest_draft.mounts is None:
+                raise InvalidAPIParameters("model vfolder id is missing on the latest revision")
             override_draft = await self._build_revision_overrides_from_spec(spec)
+            definition_path_override = spec.model_definition_path.optional_value()
+            draft_with_definition_path_override = RevisionDraft(
+                mounts=MountMetadata(
+                    model_vfolder_id=latest_draft.mounts.model_vfolder_id,
+                    model_definition_path=definition_path_override,
+                    model_mount_destination=latest_draft.mounts.model_mount_destination,
+                    extra_mounts=list(latest_draft.mounts.extra_mounts),
+                    vfolder_subpath=latest_draft.mounts.vfolder_subpath,
+                )
+            )
+            override_draft = override_draft.merge(draft_with_definition_path_override)
             # Legacy ``ModifyEndpoint`` semantics preserve untouched fields by
             # layering overrides on top of the existing revision. The
             # controller's ``add_revision`` no longer accepts a caller-provided
             # ``base`` layer, so we pre-merge here and pass the result as the
             # request-level override draft.
             overrides = latest_draft.merge(override_draft)
 
-            vfolder_id = latest_rev.model_mount_config.vfolder_id
-            if vfolder_id is None:
-                raise InvalidAPIParameters("model vfolder id is missing on the latest revision")
-            definition_path_override = spec.model_definition_path.optional_value()
-            mounts = MountMetadata(
-                model_vfolder_id=vfolder_id,
-                model_definition_path=(
-                    definition_path_override
-                    if definition_path_override is not None
-                    else latest_rev.model_mount_config.definition_path
-                ),
-                model_mount_destination=latest_rev.model_mount_config.mount_destination
-                or "/models",
-                # Carry over the latest revision's extra mounts so that an
-                # otherwise unrelated modify (e.g. replica-count) does not
-                # strip them; ``mount_perm`` survives because the
-                # revision-side projection uses the same ``MountInfoEntry``
-                # shape.
-                extra_mounts=list(latest_rev.model_mount_config.extra_mounts),
-            )
             revision = await self._deployment_controller.add_revision(
                 endpoint_id=action.deployment_id,
                 overrides=overrides,
-                mounts=mounts,
             )
             await self._deployment_controller.activate_revision(action.deployment_id, revision.id)
         elif spec.replica_count_modified():