refactor(BA-5650-G): resolve owner_id via current_user() in services

- Drop ``owner_id`` from 21 read/control session action dataclasses;
  keep it only on ``create_from_template`` / ``create_from_params`` /
  ``create_cluster`` (the three delegation-capable creation paths).
- ``services/session/service.py``: add ``_requester_user_id()`` that
  pulls the caller's UUID from the ``current_user()`` context var,
  replacing 21 ``owner_id = action.owner_id`` sites.
- ``services/session/lifecycle.py``: inject ``UserRepository`` via the
  constructor instead of instantiating it internally.
- ``registry.py``: accept ``UserRepository`` and forward it into
  ``SessionLifecycleManager``.
- ``dependencies/agents/{registry,composer}.py``: wire the repository
  through the dependency graph.
- ``repositories/user/{repository,db_source/db_source}.py``: add
  ``get_main_access_key_by_id`` (renamed from ``_by_uuid``).
- Minor adapter/service touch-ups in ``services/export`` and
  ``repositories/model_serving``.

Author: jopemachine

changes/BA-5650-G.misc.md

@@ -0,0 +1 @@
+Drop `owner_id` from read/control session action dataclasses; keep it only on creation actions (`create_from_template`, `create_from_params`, `create_cluster`). Service layer resolves the requester's UUID via `current_user()` context var. `SessionLifecycleManager` receives `UserRepository` via constructor injection instead of constructing it internally.

src/ai/backend/manager/dependencies/agents/composer.py

@@ -20,6 +20,7 @@
 from ai.backend.manager.registry import AgentRegistry
 from ai.backend.manager.repositories.deployment.repository import DeploymentRepository
 from ai.backend.manager.repositories.scheduler.repository import SchedulerRepository
+from ai.backend.manager.repositories.user.repository import UserRepository
 from ai.backend.manager.sokovan.deployment.deployment_controller import DeploymentController
 from ai.backend.manager.sokovan.deployment.revision_generator.registry import (
     RevisionGeneratorRegistry,
@@ -188,6 +189,7 @@ async def compose(
                 hook_plugin_ctx=setup_input.hook_plugin_ctx,
                 network_plugin_ctx=setup_input.network_plugin_ctx,
                 scheduling_controller=scheduling_controller,
+                user_repository=UserRepository(setup_input.db),
                 debug=setup_input.config_provider.config.debug.enabled,
                 manager_public_key=setup_input.agent_cache.manager_public_key,
                 manager_secret_key=setup_input.agent_cache.manager_secret_key,

src/ai/backend/manager/dependencies/agents/registry.py

@@ -19,6 +19,7 @@
 from ai.backend.manager.models.utils import ExtendedAsyncSAEngine
 from ai.backend.manager.plugin.network import NetworkPluginContext
 from ai.backend.manager.registry import AgentRegistry
+from ai.backend.manager.repositories.user.repository import UserRepository
 from ai.backend.manager.sokovan.scheduling_controller.scheduling_controller import (
     SchedulingController,
 )
@@ -41,6 +42,7 @@ class AgentRegistryInput:
     hook_plugin_ctx: HookPluginContext
     network_plugin_ctx: NetworkPluginContext
     scheduling_controller: SchedulingController
+    user_repository: UserRepository
     debug: bool
     manager_public_key: PublicKey
     manager_secret_key: SecretKey
@@ -82,6 +84,7 @@ async def provide(self, setup_input: AgentRegistryInput) -> AsyncIterator[AgentR
             setup_input.hook_plugin_ctx,
             setup_input.network_plugin_ctx,
             setup_input.scheduling_controller,
+            setup_input.user_repository,
             debug=setup_input.debug,
             manager_public_key=setup_input.manager_public_key,
             manager_secret_key=setup_input.manager_secret_key,

src/ai/backend/manager/registry.py

@@ -116,6 +116,7 @@
 from ai.backend.manager.plugin.network import NetworkPluginContext
 from ai.backend.manager.repositories.resource_slot import ResourceSlotRepository
 from ai.backend.manager.repositories.scheduler.types.session_creation import SessionCreationSpec
+from ai.backend.manager.repositories.user.repository import UserRepository
 from ai.backend.manager.sokovan.scheduling_controller import SchedulingController
 
 from .agent_cache import AgentRPCCache
@@ -221,6 +222,7 @@ def __init__(
         hook_plugin_ctx: HookPluginContext,
         network_plugin_ctx: NetworkPluginContext,
         scheduling_controller: SchedulingController,
+        user_repository: UserRepository,
         *,
         debug: bool = False,
         manager_public_key: PublicKey,
@@ -252,6 +254,7 @@ def __init__(
             event_producer,
             hook_plugin_ctx,
             self,
+            user_repository,
         )
         self._client_pool = ClientPool(tcp_client_session_factory)
 
@@ -460,7 +463,7 @@ async def create_session(
                 sess = await SessionRow.get_session(
                     db_session,
                     session_name,
-                    owner_access_key,
+                    owner_id=user_scope.user_uuid,
                     kernel_loading_strategy=KernelLoadingStrategy.MAIN_KERNEL_ONLY,
                 )
                 if sess.main_kernel.image is None:
@@ -687,7 +690,7 @@ async def create_cluster(
                 await SessionRow.get_session(
                     db_sess,
                     session_name,
-                    owner_access_key,
+                    owner_id=user_scope.user_uuid,
                 )
         except SessionNotFound:
             pass

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

@@ -5,7 +5,6 @@
 
 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
@@ -40,7 +39,7 @@
     UserData,
 )
 from ai.backend.manager.data.permission.types import RBACElementRef
-from ai.backend.manager.data.vfolder.types import VFolderLocation, VFolderOwnershipType
+from ai.backend.manager.data.vfolder.types import 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
@@ -81,9 +80,6 @@
     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 (
@@ -738,7 +734,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, None, kernel_loading_strategy=kernel_loading_strategy
+                    session, session_id, kernel_loading_strategy=kernel_loading_strategy
                 )
             except NoResultFound:
                 return None
@@ -832,15 +828,6 @@ 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()
@@ -873,7 +860,6 @@ 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()
@@ -955,51 +941,6 @@ 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/actions/commit_session.py

@@ -2,7 +2,6 @@
 from dataclasses import dataclass
 from typing import Any, override
 
-from ai.backend.common.types import AccessKey
 from ai.backend.manager.actions.action import BaseActionResult
 from ai.backend.manager.actions.types import ActionOperationType
 from ai.backend.manager.data.session.types import SessionData
@@ -12,7 +11,6 @@
 @dataclass
 class CommitSessionAction(SessionCommitAction):
     session_name: str
-    owner_access_key: AccessKey
     filename: str | None
 
     @override

src/ai/backend/manager/services/session/actions/complete.py

@@ -3,7 +3,6 @@
 from typing import Any, override
 
 from ai.backend.common.dto.agent.response import CodeCompletionResp
-from ai.backend.common.types import AccessKey
 from ai.backend.manager.actions.action import BaseActionResult
 from ai.backend.manager.actions.types import ActionOperationType
 from ai.backend.manager.data.session.types import SessionData
@@ -14,7 +13,6 @@
 @dataclass
 class CompleteAction(SessionAction):
     session_name: str
-    owner_access_key: AccessKey
     code: str
     # TODO: Add type
     options: Mapping[str, Any] | None

src/ai/backend/manager/services/session/actions/convert_session_to_image.py

@@ -3,7 +3,6 @@
 from typing import override
 
 from ai.backend.common.data.session.types import CustomizedImageVisibilityScope
-from ai.backend.common.types import AccessKey
 from ai.backend.manager.actions.action import BaseActionResult
 from ai.backend.manager.actions.types import ActionOperationType
 from ai.backend.manager.data.session.types import SessionData
@@ -13,7 +12,6 @@
 @dataclass
 class ConvertSessionToImageAction(SessionAction):
     session_name: str
-    owner_access_key: AccessKey
     image_name: str
     image_visibility: CustomizedImageVisibilityScope
     image_owner_id: uuid.UUID

src/ai/backend/manager/services/session/actions/create_cluster.py

@@ -30,7 +30,7 @@ class CreateClusterAction(SessionScopeAction):
     domain_name: str
     scaling_group_name: str
     requester_access_key: AccessKey
-    owner_access_key: AccessKey
+    owner_id: uuid.UUID
     tag: str
     enqueue_only: bool
     keypair_resource_policy: dict[str, Any] | None

src/ai/backend/manager/services/session/actions/create_from_params.py

@@ -30,7 +30,7 @@ class CreateFromParamsActionParams:
     tag: str
     priority: int
     is_preemptible: bool
-    owner_access_key: AccessKey
+    owner_id: uuid.UUID
     enqueue_only: bool
     max_wait_seconds: int
     starts_at: str | None