refactor(BA-5715): address slice G review feedback

- Revert stray changes in repositories/model_serving/repository.py
  (DeploymentStorageSource import / model-definition fetch removal
  were unrelated to BA-5650).
- services/session/service.py: _resolve_owner_main_access_key now
  uses the narrower UserRepository.get_main_access_key_by_id helper
  instead of loading the entire UserData record.
- services/session/lifecycle.py: guard the POST_START_SESSION hook
  payload against a None main_access_key — log a warning and skip the
  hook rather than pass None into plugins that expect an AccessKey.

Author: jopemachine

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

@@ -5,6 +5,7 @@
 
 import sqlalchemy as sa
 from pydantic import HttpUrl
+from ruamel.yaml import YAML
 from sqlalchemy.exc import IntegrityError, NoResultFound, StatementError
 from sqlalchemy.ext.asyncio import AsyncSession as SASession
 from sqlalchemy.orm import selectinload
@@ -39,7 +40,7 @@
     UserData,
 )
 from ai.backend.manager.data.permission.types import RBACElementRef
-from ai.backend.manager.data.vfolder.types import VFolderOwnershipType
+from ai.backend.manager.data.vfolder.types import VFolderLocation, VFolderOwnershipType
 from ai.backend.manager.errors.common import ObjectNotFound
 from ai.backend.manager.errors.resource import DatabaseConnectionUnavailable
 from ai.backend.manager.errors.service import EndpointNotFound
@@ -80,6 +81,9 @@
     execute_rbac_entity_creator,
 )
 from ai.backend.manager.repositories.deployment.creators import DeploymentPolicyCreatorSpec
+from ai.backend.manager.repositories.deployment.storage_source.storage_source import (
+    DeploymentStorageSource,
+)
 from ai.backend.manager.repositories.model_serving.updaters import EndpointUpdaterSpec
 from ai.backend.manager.services.model_serving.actions.modify_endpoint import ModifyEndpointAction
 from ai.backend.manager.services.model_serving.exceptions import (
@@ -734,7 +738,7 @@ async def get_session_by_id(
         async with self._db.begin_readonly_session_read_committed() as session:
             try:
                 return await SessionRow.get_session(
-                    session, session_id, kernel_loading_strategy=kernel_loading_strategy
+                    session, session_id, None, kernel_loading_strategy=kernel_loading_strategy
                 )
             except NoResultFound:
                 return None
@@ -828,6 +832,15 @@ async def _do_mutate() -> MutationResult:
                     if current_rev is None:
                         raise InvalidAPIParameters("Endpoint has no current revision")
 
+                    # Re-read model definition from vfolder to pick up file changes
+                    refreshed_model_definition = await self._fetch_model_definition_from_vfolder(
+                        db_session,
+                        storage_manager,
+                        current_rev.model,
+                        spec.model_definition_path.optional_value()
+                        or current_rev.model_definition_path,
+                    )
+
                     # Resolve image if changed
                     image_id = current_rev.image
                     image_ref = spec.image.optional_value()
@@ -860,6 +873,7 @@ async def _do_mutate() -> MutationResult:
                             if spec.model_definition_path.optional_value() is not None
                             else current_rev.model_definition_path
                         ),
+                        model_definition=refreshed_model_definition or current_rev.model_definition,
                         resource_group=endpoint_row.resource_group,
                         resource_opts=(
                             spec.resource_opts.optional_value()
@@ -941,6 +955,51 @@ async def _do_mutate() -> MutationResult:
         except Exception:
             raise
 
+    async def _fetch_model_definition_from_vfolder(
+        self,
+        db_session: SASession,
+        storage_manager: StorageSessionManager,
+        vfolder_id: uuid.UUID | None,
+        model_definition_path: str | None,
+    ) -> dict[str, Any] | None:
+        """Re-read model definition file from the vfolder storage.
+
+        Returns the parsed YAML content, or None if the file cannot be read.
+        """
+        if vfolder_id is None:
+            return None
+        try:
+            vf_query = sa.select(
+                VFolderRow.id,
+                VFolderRow.host,
+                VFolderRow.quota_scope_id,
+                VFolderRow.ownership_type,
+                VFolderRow.usage_mode,
+            ).where(VFolderRow.id == vfolder_id)
+            vf_result = await db_session.execute(vf_query)
+            vf_row = vf_result.one_or_none()
+            if vf_row is None:
+                return None
+
+            vfolder_location = VFolderLocation(
+                id=vf_row.id,
+                quota_scope_id=vf_row.quota_scope_id,
+                host=vf_row.host,
+                ownership_type=vf_row.ownership_type,
+                usage_mode=vf_row.usage_mode,
+            )
+            candidates = (
+                [model_definition_path]
+                if model_definition_path
+                else ["model-definition.yaml", "model-definition.yml"]
+            )
+            storage_source = DeploymentStorageSource(storage_manager)
+            content = await storage_source.fetch_definition_file(vfolder_location, candidates)
+            yaml = YAML()
+            return cast(dict[str, Any], yaml.load(content))
+        except Exception:
+            return None
+
     @model_serving_repository_resilience.apply()
     async def search_auto_scaling_rules(
         self,

src/ai/backend/manager/services/session/lifecycle.py

@@ -23,6 +23,7 @@
 )
 from ai.backend.common.plugin.hook import HookPluginContext
 from ai.backend.common.types import (
+    AccessKey,
     ResourceSlot,
     SessionId,
     SessionTypes,
@@ -136,18 +137,29 @@ async def _post_status_transition(
                     SessionStartedAnycastEvent(session_row.id, creation_id)
                 )
                 # BA-5609: resolve main_access_key from owner_id; external
-                # hook plugins still receive the resolved access key.
+                # hook plugins still receive the resolved access key. If the
+                # owner has no main_access_key configured, skip the hook —
+                # calling it with ``None`` would likely break plugins that
+                # assume a non-null keypair identifier.
                 session_main_access_key = await self._user_repository.get_main_access_key_by_id(
                     session_row.user_uuid
                 )
-                await self.hook_plugin_ctx.notify(
-                    "POST_START_SESSION",
-                    (
+                if session_main_access_key is not None:
+                    await self.hook_plugin_ctx.notify(
+                        "POST_START_SESSION",
+                        (
+                            session_row.id,
+                            session_row.name,
+                            AccessKey(session_main_access_key),
+                        ),
+                    )
+                else:
+                    log.warning(
+                        "POST_START_SESSION skipped: owner {} has no main_access_key"
+                        " (session {})",
+                        session_row.user_uuid,
                         session_row.id,
-                        session_row.name,
-                        session_main_access_key,
-                    ),
-                )
+                    )
                 match session_row.session_type:
                     case SessionTypes.BATCH:
                         await self.registry.trigger_batch_execution(session_row)

src/ai/backend/manager/services/session/service.py

@@ -288,16 +288,17 @@ async def _resolve_owner_main_access_key(
     ) -> AccessKey:
         """Resolve a delegated owner UUID to that user's main access key.
 
-        Loads the target user via the user repository and returns the main
-        access key. Raises ``InternalServerError`` if the target user has no
-        main access key configured.
+        Uses the narrower ``UserRepository.get_main_access_key_by_id`` helper
+        so we only fetch the single scalar column we need. Raises
+        ``InternalServerError`` if the target user has no main access key
+        configured.
         """
-        user_data = await self._user_repository.get_user_by_uuid(owner_id)
-        if user_data.main_access_key is None:
+        main_access_key = await self._user_repository.get_main_access_key_by_id(owner_id)
+        if main_access_key is None:
             raise InternalServerError(
                 f"Delegated owner {owner_id} has no main access key configured"
             )
-        return AccessKey(user_data.main_access_key)
+        return AccessKey(main_access_key)
 
     async def commit_session(self, action: CommitSessionAction) -> CommitSessionActionResult:
         session_name = action.session_name