fix(BA-5650-I): resolve cascaded rebase conflicts and align tests

- Resolve conflict markers in scheduler db_source by keeping the
  user-table-joined ``main_access_key`` lookup (single source of truth)
  rather than the obsolete ``session.access_key`` snapshot.
- Drop duplicate ``resolve_main_access_keys`` definition introduced
  by the rebase.
- Use ``spec.access_key`` (SessionCreationSpec) where applicable —
  the spec dataclass keeps ``access_key`` for now.
- Test fixture ``_make_session_data`` now passes ``owner_id`` instead
  of relying on attribute assignment after construction.
- Use ``main_access_key`` on PendingSessionData in provisioner test.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: jopemachine

src/ai/backend/manager/repositories/scheduler/db_source/db_source.py

@@ -1215,8 +1215,7 @@ async def get_pending_timeout_sessions_by_ids(
                         SweptSessionInfo(
                             session_id=row.id,
                             creation_id=row.creation_id,
-                            main_access_key=row.access_key,
->>>>>>> e8b82576d (refactor(BA-5650-I): test and remaining ORM updates)
+                            main_access_key=row.main_access_key,
                         )
                     )
 
@@ -1489,7 +1488,7 @@ async def fetch_session_creation_data(
                 db_sess,
                 spec.user_scope.domain_name,
                 str(spec.user_scope.group_id),
-                spec.main_access_key,
+                spec.access_key,
             )
             image_infos = await self._resolve_image_info(db_sess, image_refs)
 
@@ -1510,7 +1509,7 @@ async def fetch_session_creation_data(
             dotfile_data = await self._fetch_dotfiles(
                 db_sess,
                 spec.user_scope,
-                spec.main_access_key,
+                spec.access_key,
                 vfolder_mounts,
             )
 
@@ -1554,7 +1553,7 @@ async def fetch_session_creation_context(
                 db_sess,
                 spec.user_scope.domain_name,
                 str(spec.user_scope.group_id),
-                spec.main_access_key,
+                spec.access_key,
             )
             image_infos = await self._resolve_image_info(db_sess, image_refs)
 
@@ -2927,12 +2926,7 @@ async def _get_sessions_by_statuses(
             scheduled_session = ScheduledSessionData(
                 session_id=session.id,
                 creation_id=session.creation_id or "",
-                main_access_key=AccessKey(session.access_key)
-                if session.access_key
-                else AccessKey(""),
-=======
                 main_access_key=AccessKey(owner_main_ak) if owner_main_ak else AccessKey(""),
->>>>>>> 4ac120c73 (refactor(BA-5650-I): drop stray non-BA-5650 changes from slice)
                 reason="triggered-by-scheduler",
             )
             scheduled_sessions.append(scheduled_session)
@@ -2979,12 +2973,7 @@ async def _get_scheduled_sessions(self, db_sess: SASession) -> list[ScheduledSes
                 ScheduledSessionData(
                     session_id=session.id,
                     creation_id=session.creation_id or "",
-                    main_access_key=AccessKey(session.access_key)
-                    if session.access_key
-                    else AccessKey(""),
-=======
                     main_access_key=AccessKey(owner_main_ak) if owner_main_ak else AccessKey(""),
->>>>>>> 4ac120c73 (refactor(BA-5650-I): drop stray non-BA-5650 changes from slice)
                     reason="triggered-by-scheduler",
                 )
             )
@@ -4828,21 +4817,3 @@ async def _get_db_now_in_session(self, db_sess: SASession) -> datetime:
         """
         result = await db_sess.execute(sa.select(sa.func.now()))
         return result.scalar_one()
-
-    async def resolve_main_access_keys(
-        self, session_ids: Sequence[SessionId]
-    ) -> dict[SessionId, AccessKey]:
-        """Resolve the main access key for each session's owner.
-
-        Joins sessions → users to look up the owner's main_access_key.
-        """
-        if not session_ids:
-            return {}
-        async with self._db.begin_readonly_session() as db_sess:
-            stmt = (
-                sa.select(SessionRow.id, UserRow.main_access_key)
-                .join(UserRow, SessionRow.user_uuid == UserRow.uuid)
-                .where(SessionRow.id.in_([sid for sid in session_ids]))
-            )
-            rows = (await db_sess.execute(stmt)).all()
-            return {SessionId(row[0]): AccessKey(row[1]) for row in rows if row[1] is not None}

tests/unit/manager/services/session/test_session_lifecycle_service.py

@@ -210,6 +210,7 @@ def _make_session_data(
 ) -> SessionData:
     return SessionData(
         id=session_id,
+        owner_id=user_id,
         creation_id="test-creation-id",
         name=name,
         session_type=session_type,

tests/unit/manager/services/session/test_session_service.py

@@ -245,10 +245,7 @@ def sample_session_data(
         agent_ids=["i-ubuntu"],
         domain_name="default",
         group_id=sample_group_id,
-<<<<<<< HEAD
         owner_id=sample_user_id,
-=======
->>>>>>> d59487b17 (refactor(BA-5650-I): test and remaining ORM updates)
         images=["cr.backend.ai/stable/python:latest"],
         tag=None,
         occupying_slots=ResourceSlot({"cpu": 1, "mem": 1024}),
@@ -349,10 +346,7 @@ async def test_multiple_matches(
                 agent_ids=[],
                 domain_name="default",
                 group_id=sample_group_id,
-<<<<<<< HEAD
                 owner_id=sample_user_id,
-=======
->>>>>>> d59487b17 (refactor(BA-5650-I): test and remaining ORM updates)
                 images=["python:latest"],
                 tag=None,
                 occupying_slots=ResourceSlot({}),

tests/unit/manager/sokovan/scheduler/provisioner/test_provisioner.py

@@ -66,7 +66,7 @@ def _create_scheduling_data_with_strategy(
     # Create one pending session
     session = PendingSessionData(
         id=SessionId(uuid.uuid4()),
-        access_key=AccessKey("test-key"),
+        main_access_key=AccessKey("test-key"),
         requested_slots=ResourceSlot({"cpu": Decimal("1"), "mem": Decimal("1024")}),
         owner_id=uuid.uuid4(),
         group_id=uuid.uuid4(),