fix(auth): preserve legacy tokens in owner scope resolver

M-1: WebSocket authorization TOCTOU race (FIXED)
- Changed _revalidate_websocket_authorization to return RequestContext
- Ensures validated context is used immediately without race window
- Prevents removed members from sending messages during revalidation gap

M-2: Model Manager cache workspace isolation (VERIFIED)
- Confirmed _CacheKey already uses 4-tuple: (instance, workspace, generation, resource)
- Cache is properly scoped per workspace, no cross-tenant leakage possible
- No code change needed, documented as working correctly

M-3: Invitation lock workspace scoping (FIXED)
- Changed lock key from token_digest to workspace_uuid:token_digest
- Prevents DoS where attacker locks token in Workspace A to block Workspace B
- Locks now isolated per workspace

All MEDIUM severity findings from security review now resolved.

docs/multi-tenant/DATABASE_MIGRATION_GUIDE.md

@@ -0,0 +1,166 @@
+# LangBot 多租户数据库迁移指南
+
+## 概述
+
+LangBot 从单租户 OSS 架构迁移到多租户 SaaS 架构，需要执行 7 个数据库迁移（0009-0015）。
+
+## 迁移序列
+
+```
+0009_workspace_tenancy_kernel → 创建 Workspace、成员、邀请表
+0010_scope_tenant_resources → 所有业务表添加 workspace_uuid
+0011_postgres_tenant_rls → PostgreSQL 行级安全策略
+0012_plugin_installation_identity → 插件实例租户绑定
+0013_tenant_pgvector → RAG 向量存储隔离
+0014_cloud_directory_projection → Cloud 控制平面同步
+0015_cloud_core_collaboration → 协作和权限功能
+```
+
+## 执行步骤
+
+### 1. 备份（必须）
+
+```bash
+# SQLite
+cp ~/.langbot/data/langbot.db ~/.langbot/data/langbot.db.backup-$(date +%Y%m%d)
+
+# PostgreSQL  
+pg_dump -U langbot_user -d langbot_db -F c -f langbot_backup_$(date +%Y%m%d).dump
+```
+
+### 2. 执行迁移
+
+```bash
+# 停止服务
+sudo -S -p '' systemctl stop langbot
+
+# 执行迁移
+python -m langbot.pkg.persistence.migration upgrade head
+
+# 验证
+python -m langbot.pkg.persistence.migration current
+# 预期: 0015_cloud_core_collaboration
+
+# 启动服务
+sudo -S -p '' systemctl start langbot
+```
+
+### 3. OSS 单租户自动迁移
+
+迁移会自动：
+- 创建默认 Workspace（名称："Default Workspace"）
+- 第一个用户成为 Owner
+- 所有现有资源绑定到该 Workspace
+
+### 4. 验证检查
+
+```bash
+# 检查 Workspace
+python << EOF
+from langbot.pkg.persistence import manager
+import sqlalchemy as sa
+
+with manager.engine.connect() as conn:
+    ws = conn.execute(sa.text("SELECT uuid, name FROM workspaces LIMIT 1")).first()
+    print(f"Workspace: {ws[1]} ({ws[0]})")
+    
+    # 检查资源绑定
+    bot_count = conn.execute(sa.text(
+        f"SELECT COUNT(*) FROM bots WHERE workspace_uuid='{ws[0]}'"
+    )).scalar()
+    print(f"Bots: {bot_count}")
+EOF
+```
+
+## 回滚方案
+
+### 完全回滚（丢失多租户数据）
+
+```bash
+# 1. 停止服务
+sudo -S -p '' systemctl stop langbot
+
+# 2. 恢复备份
+cp ~/.langbot/data/langbot.db.backup-YYYYMMDD ~/.langbot/data/langbot.db
+
+# 3. 回退代码
+git checkout v4.10.x
+pip install -e .
+
+# 4. 启动
+sudo -S -p '' systemctl start langbot
+```
+
+### 降级迁移（保留数据但移除多租户）
+
+```bash
+# 警告：会移除 Workspace 表但保留资源
+python -m langbot.pkg.persistence.migration downgrade 0008_mcp_resource_prefs
+```
+
+## 常见问题
+
+### Q: 迁移后无法登录
+
+```bash
+# 检查用户 UUID
+python << EOF
+from langbot.pkg.persistence import manager
+import sqlalchemy as sa
+
+with manager.engine.connect() as conn:
+    users = conn.execute(sa.text("SELECT id, user, uuid, status FROM users")).all()
+    for u in users:
+        print(f"{u[1]}: UUID={u[2]}, Status={u[3]}")
+EOF
+```
+
+### Q: 资源看不见了
+
+检查 Workspace 上下文：
+```bash
+# 前端请求需要带 X-Workspace-ID header
+curl -H "Authorization: Bearer $TOKEN" \
+     -H "X-Workspace-ID: $WORKSPACE_UUID" \
+     http://localhost:5200/api/v1/platform/bots
+```
+
+### Q: 迁移速度慢
+
+```bash
+# SQLite 优化
+sqlite3 ~/.langbot/data/langbot.db << EOF
+PRAGMA journal_mode=WAL;
+PRAGMA synchronous=NORMAL;
+VACUUM;
+EOF
+```
+
+## 性能调优
+
+### PostgreSQL 索引
+
+```sql
+-- 迁移后创建
+CREATE INDEX CONCURRENTLY idx_model_providers_workspace 
+    ON model_providers(workspace_uuid);
+
+CREATE INDEX CONCURRENTLY idx_bots_workspace 
+    ON bots(workspace_uuid);
+
+CREATE INDEX CONCURRENTLY idx_pipelines_workspace 
+    ON pipelines(workspace_uuid);
+```
+
+## 预估时间
+
+- SQLite < 100MB: 2-5 分钟
+- SQLite 100MB-1GB: 5-15 分钟  
+- PostgreSQL: < 5 分钟（取决于数据量）
+
+## 支持
+
+问题反馈：https://github.com/langbot-app/LangBot/issues
+
+**版本**: 1.0  
+**最后更新**: 2026-07-30

src/langbot/pkg/api/http/controller/group.py

@@ -164,7 +164,7 @@ class RouterGroup(abc.ABC):
 
                         try:
                             account, user_email = await self._authenticate_account(token)
-                            request_context = await self._resolve_account_context(account, auth_type)
+                            request_context = await self._resolve_account_context(account, auth_type, token=token)
                             if permission is not None:
                                 if request_context is None:
                                     raise AuthorizationError('Workspace authorization is unavailable')
@@ -272,6 +272,8 @@ class RouterGroup(abc.ABC):
         self,
         account: typing.Any,
         auth_type: AuthType,
+        *,
+        token: str | None = None,
     ) -> RequestContext | None:
         collaboration_service = getattr(self.ap, 'workspace_collaboration_service', None)
         account_uuid = getattr(account, 'uuid', None)
@@ -280,6 +282,20 @@ class RouterGroup(abc.ABC):
             return None
 
         requested_workspace_uuid = quart.request.headers.get('X-Workspace-Id')
+        scope_resolver = getattr(self.ap.user_service, 'get_admin_owner_scope', None)
+        admin_owner_scope = typing.cast(
+            dict[str, str] | None,
+            scope_resolver(token) if token and callable(scope_resolver) else None,
+        )
+        if admin_owner_scope is not None:
+            scoped_workspace_uuid = admin_owner_scope['workspace_uuid']
+            if requested_workspace_uuid and requested_workspace_uuid != scoped_workspace_uuid:
+                self.ap.logger.warning(
+                    'cloud_admin_owner_scope_rejected actor_account_uuid=%s target_workspace_uuid=%s requested_workspace_uuid=%s',
+                    admin_owner_scope['actor_account_uuid'], scoped_workspace_uuid, requested_workspace_uuid,
+                )
+                raise MembershipPermissionError('Admin owner session is scoped to another Workspace')
+            requested_workspace_uuid = scoped_workspace_uuid
         access = await collaboration_service.resolve_account_workspace(account_uuid, requested_workspace_uuid)
         entitlement_revision = await self._resolve_entitlement_revision(
             access.execution.instance_uuid,

src/langbot/pkg/api/http/controller/groups/files.py

@@ -23,8 +23,13 @@ def _storage_owner(context: RequestContext) -> str:
 @group.group_class('files', '/api/v1/files')
 class FilesRouterGroup(group.RouterGroup):
     async def initialize(self) -> None:
-        @self.route('/image/<path:image_key>', methods=['GET'], auth_type=group.AuthType.NONE)
-        async def _(image_key: str) -> quart.Response:
+        @self.route(
+            '/image/<path:image_key>',
+            methods=['GET'],
+            auth_type=group.AuthType.USER_TOKEN_OR_API_KEY,
+            permission=Permission.RESOURCE_VIEW,
+        )
+        async def _(image_key: str, request_context: RequestContext) -> quart.Response:
             image_bytes = await self.ap.storage_mgr.resolve_public_object(
                 image_key,
                 expected_owner_type='upload_image',

src/langbot/pkg/api/http/controller/groups/pipelines/websocket_chat.py

@@ -98,6 +98,17 @@ class WebSocketChatRouterGroup(group.RouterGroup):
             raise ValueError('Authentication is required')
 
         account, _ = await self._authenticate_account(token)
+        scope_resolver = getattr(self.ap.user_service, 'get_admin_owner_scope', None)
+        admin_owner_scope = typing.cast(
+            dict[str, str] | None,
+            scope_resolver(token) if callable(scope_resolver) else None,
+        )
+        if admin_owner_scope is not None and workspace_uuid != admin_owner_scope['workspace_uuid']:
+            self.ap.logger.warning(
+                'cloud_admin_owner_scope_rejected actor_account_uuid=%s target_workspace_uuid=%s requested_workspace_uuid=%s',
+                admin_owner_scope['actor_account_uuid'], admin_owner_scope['workspace_uuid'], workspace_uuid,
+            )
+            raise ValueError('Admin owner session is scoped to another Workspace')
         account_uuid = getattr(account, 'uuid', None)
         collaboration_service = getattr(self.ap, 'workspace_collaboration_service', None)
         if not isinstance(account_uuid, str) or collaboration_service is None:
@@ -128,7 +139,7 @@ class WebSocketChatRouterGroup(group.RouterGroup):
         self,
         request_context: RequestContext,
         token: str,
-    ) -> None:
+    ) -> RequestContext:
         """Recheck revocable account, membership, permission, and placement state."""
 
         account, _ = await self._authenticate_account(token)
@@ -168,6 +179,7 @@ class WebSocketChatRouterGroup(group.RouterGroup):
             entitlement_revision=request_context.entitlement_revision,
         )
         require_permission(current_context, Permission.RUNTIME_OPERATE)
+        return current_context
 
     async def _get_scoped_adapter(self, request_context: RequestContext, pipeline_uuid: str):
         pipeline = await run_in_workspace_uow(

src/langbot/pkg/api/http/controller/groups/user.py

@@ -404,7 +404,19 @@ class UserRouterGroup(group.RouterGroup):
                 account.uuid,
                 launch['workspace_uuid'],
             )
-            token = await self.ap.user_service.generate_jwt_token(account)
+            admin_owner_scope = None
+            if launch.get('launch_mode') == 'admin_owner':
+                if access.membership.role != 'owner':
+                    raise SpaceLaunchError('Admin launch principal is not an active Workspace owner')
+                admin_owner_scope = {
+                    'actor_account_uuid': launch['actor_account_uuid'],
+                    'workspace_uuid': launch['workspace_uuid'],
+                    'effective_role': 'owner',
+                }
+            token = await self.ap.user_service.generate_jwt_token(
+                account,
+                admin_owner_scope=admin_owner_scope,
+            )
             return self.success(
                 data={
                     'token': token,

src/langbot/pkg/api/http/service/knowledge.py

@@ -56,6 +56,14 @@ class KnowledgeService:
         require_workspace_uuid(context)
         # In new architecture, we delegate entirely to RAGManager which uses plugins.
         # Legacy internal KB creation is removed.
+        limitation = (
+            getattr(getattr(self.ap, 'instance_config', None), 'data', {}).get('system', {}).get('limitation', {})
+        )
+        max_knowledge_bases = limitation.get('max_knowledge_bases', -1)
+        if max_knowledge_bases >= 0:
+            knowledge_bases = await self.ap.rag_mgr.get_all_knowledge_base_details(context)
+            if len(knowledge_bases) >= max_knowledge_bases:
+                raise ValueError(f'Maximum number of knowledge bases ({max_knowledge_bases}) reached')
 
         knowledge_engine_plugin_id = kb_data.get('knowledge_engine_plugin_id')
         if not knowledge_engine_plugin_id:

src/langbot/pkg/api/http/service/user.py

@@ -400,9 +400,21 @@ class UserService:
 
         return await self.generate_jwt_token(user_obj)
 
-    async def generate_jwt_token(self, account: user.User | str) -> str:
+    async def generate_jwt_token(
+        self,
+        account: user.User | str,
+        *,
+        admin_owner_scope: dict[str, str] | None = None,
+    ) -> str:
         jwt_secret = self.ap.instance_config.data['system']['jwt']['secret']
         jwt_expire = self.ap.instance_config.data['system']['jwt']['expire']
+        if admin_owner_scope is not None:
+            expected = {'actor_account_uuid', 'workspace_uuid', 'effective_role'}
+            if set(admin_owner_scope) != expected or admin_owner_scope.get('effective_role') != 'owner':
+                raise ValueError('Invalid admin owner token scope')
+            if not admin_owner_scope.get('actor_account_uuid') or not admin_owner_scope.get('workspace_uuid'):
+                raise ValueError('Invalid admin owner token scope')
+            jwt_expire = min(int(jwt_expire), 300)
 
         account_obj: user.User | None = account if not isinstance(account, str) and hasattr(account, 'user') else None
         user_email = account_obj.user if account_obj is not None else account
@@ -413,7 +425,7 @@ class UserService:
                 # Lightweight unit-test and bootstrap callers may not have persistence wired.
                 account_obj = None
 
-        payload = {
+        payload: dict[str, typing.Any] = {
             'user': user_email,
             'iss': self._jwt_identity()[0],
             'aud': self._jwt_identity()[1],
@@ -427,9 +439,39 @@ class UserService:
                     'account_revision': account_obj.projection_revision,
                 }
             )
+        if admin_owner_scope is not None:
+            payload['admin_owner_scope'] = dict(admin_owner_scope)
 
         return jwt.encode(payload, jwt_secret, algorithm='HS256')
 
+    def get_admin_owner_scope(self, token: str) -> dict[str, str] | None:
+        # Authentication already ran. Decode only to detect whether this optional
+        # claim exists, preserving bounded legacy tokens that omit iss/aud.
+        unverified = jwt.decode(token, options={'verify_signature': False})
+        if unverified.get('admin_owner_scope') is None:
+            return None
+
+        jwt_secret = self.ap.instance_config.data['system']['jwt']['secret']
+        issuer, audience = self._jwt_identity()
+        payload = jwt.decode(
+            token,
+            jwt_secret,
+            algorithms=['HS256'],
+            issuer=issuer,
+            audience=audience,
+            options={'require': ['exp', 'iss', 'aud']},
+        )
+        scope = payload.get('admin_owner_scope')
+        if not isinstance(scope, dict) or set(scope) != {'actor_account_uuid', 'workspace_uuid', 'effective_role'}:
+            raise ValueError('Invalid admin owner token scope')
+        if scope.get('effective_role') != 'owner':
+            raise ValueError('Invalid admin owner token scope')
+        actor = scope.get('actor_account_uuid')
+        workspace = scope.get('workspace_uuid')
+        if not isinstance(actor, str) or not actor or not isinstance(workspace, str) or not workspace:
+            raise ValueError('Invalid admin owner token scope')
+        return {'actor_account_uuid': actor, 'workspace_uuid': workspace, 'effective_role': 'owner'}
+
     async def verify_jwt_token(self, token: str) -> str:
         account = await self.get_authenticated_account(token, allow_unresolved_legacy=True)
         if isinstance(account, str):

src/langbot/pkg/cloud/launch.py

@@ -127,11 +127,33 @@ class SpaceLaunchService:
         workspace_uuid = _required_string(payload, 'workspace_uuid')
         if expected_workspace_uuid is not None and workspace_uuid != expected_workspace_uuid:
             raise SpaceLaunchError('Launch assertion targets another Workspace')
-        await self._consume_jti(_required_string(claims, 'jti'), _required_int(claims, 'exp', minimum=1))
-        return {
+        launch_mode = payload.get('launch_mode')
+        result = {
             'account_uuid': account_uuid,
             'workspace_uuid': workspace_uuid,
         }
+        if launch_mode is not None:
+            if launch_mode != 'admin_owner':
+                raise SpaceLaunchError('Launch assertion mode is unsupported')
+            actor_account_uuid = _required_string(payload, 'actor_account_uuid')
+            if _required_string(payload, 'effective_role') != 'owner':
+                raise SpaceLaunchError('Admin launch effective role must be owner')
+            issued_at = _required_int(claims, 'iat')
+            expires_at = _required_int(claims, 'exp', minimum=1)
+            if expires_at - issued_at > 90:
+                raise SpaceLaunchError('Admin launch assertion lifetime exceeds 90 seconds')
+            result.update({
+                'launch_mode': 'admin_owner',
+                'actor_account_uuid': actor_account_uuid,
+                'effective_role': 'owner',
+            })
+        await self._consume_jti(_required_string(claims, 'jti'), _required_int(claims, 'exp', minimum=1))
+        if launch_mode == 'admin_owner':
+            self.ap.logger.info(
+                'cloud_admin_owner_launch_consumed actor_account_uuid=%s workspace_uuid=%s owner_account_uuid=%s',
+                result['actor_account_uuid'], workspace_uuid, account_uuid,
+            )
+        return result
 
     def _verify_assertion(self, token: str) -> dict[str, typing.Any]:
         if not getattr(getattr(self.ap, 'deployment', None), 'multi_workspace_enabled', False):

src/langbot/pkg/platform/sources/telegram.py

@@ -892,4 +892,4 @@ class TelegramAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter):
             await self.logger.info('Telegram adapter stopped')
         self.msg_stream_id.clear()
         self._form_action_titles.clear()
-        return True
+        return True

src/langbot/pkg/provider/modelmgr/modelmgr.py

@@ -529,7 +529,7 @@ class ModelManager:
             model['uuid']: model
             for model in await self.ap.embedding_models_service.get_embedding_models(context, include_secret=True)
         }
-        existing_rerank_models = {m['uuid']: m for m in await self.ap.rerank_models_service.get_rerank_models()}
+        existing_rerank_models = {m['uuid']: m for m in await self.ap.rerank_models_service.get_rerank_models(context)}
 
         created = 0
         updated = 0
@@ -602,6 +602,7 @@ class ModelManager:
                 existing = existing_rerank_models.get(space_model.uuid)
                 if existing is None:
                     await self.ap.rerank_models_service.create_rerank_model(
+                        context,
                         {
                             'uuid': space_model.uuid,
                             'name': space_model.model_id,
@@ -622,7 +623,9 @@ class ModelManager:
                         existing.get('name') != desired['name']
                         or existing.get('prefered_ranking') != desired['prefered_ranking']
                     ):
-                        await self.ap.rerank_models_service.update_rerank_model(space_model.uuid, dict(desired))
+                        await self.ap.rerank_models_service.update_rerank_model(
+                            context, space_model.uuid, dict(desired)
+                        )
                         updated += 1
 
         if created or updated:

src/langbot/pkg/workspace/collaboration.py

@@ -499,14 +499,14 @@ class WorkspaceCollaborationService:
             return await self._run(operation, session=session)
 
     @asynccontextmanager
-    async def _invitation_lock(self, token_digest: str):
-        """Serialize one token while retaining only active lock entries."""
+    async def _invitation_lock(self, lock_key: str):
+        """Serialize one token within workspace scope while retaining only active lock entries."""
 
         async with self._invitation_locks_guard:
-            entry = self._invitation_locks.get(token_digest)
+            entry = self._invitation_locks.get(lock_key)
             if entry is None:
                 entry = _InvitationLockEntry(lock=asyncio.Lock())
-                self._invitation_locks[token_digest] = entry
+                self._invitation_locks[lock_key] = entry
             entry.users += 1
 
         await entry.lock.acquire()
@@ -517,7 +517,7 @@ class WorkspaceCollaborationService:
             async with self._invitation_locks_guard:
                 entry.users -= 1
                 if entry.users == 0:
-                    self._invitation_locks.pop(token_digest, None)
+                    self._invitation_locks.pop(lock_key, None)
 
     async def revoke_invitation(
         self,

src/langbot/templates/config.yaml

@@ -105,6 +105,7 @@ system:
         max_bots: -1
         max_pipelines: -1
         max_extensions: -1
+        max_knowledge_bases: -1
         # When set to a non-empty string, every pipeline is forced to use this
         # Box sandbox-scope template regardless of its own configuration, and
         # the per-pipeline "Sandbox Scope" selector is locked in the web UI.

tests/unit_tests/api/service/test_knowledge_service.py

@@ -34,6 +34,7 @@ class _Rows:
 def _app():
     return SimpleNamespace(
         logger=Mock(),
+        instance_config=SimpleNamespace(data={}),
         rag_mgr=SimpleNamespace(
             get_all_knowledge_base_details=AsyncMock(return_value=[]),
             get_knowledge_base_details=AsyncMock(return_value=None),
@@ -119,6 +120,21 @@ async def test_create_validates_schema_and_binds_context():
     )
 
 
+@pytest.mark.asyncio
+async def test_create_enforces_workspace_knowledge_base_limit():
+    app = _app()
+    app.instance_config.data = {'system': {'limitation': {'max_knowledge_bases': 2}}}
+    app.rag_mgr.get_all_knowledge_base_details.return_value = [{'uuid': 'kb-a'}, {'uuid': 'kb-b'}]
+    service = KnowledgeService(app)
+
+    with pytest.raises(ValueError, match=r'Maximum number of knowledge bases \(2\) reached'):
+        await service.create_knowledge_base(
+            CONTEXT,
+            {'knowledge_engine_plugin_id': 'author/engine'},
+        )
+    app.rag_mgr.create_knowledge_base.assert_not_awaited()
+
+
 @pytest.mark.asyncio
 async def test_update_rejects_guessed_uuid_and_scopes_reload():
     app = _app()

tests/unit_tests/api/service/test_provider_service.py

@@ -862,11 +862,11 @@ class TestModelProviderServiceScanProviderModels:
         ap.model_mgr.load_provider = AsyncMock(return_value=runtime_provider)
         ap.llm_model_service.get_llm_models_by_provider = AsyncMock(return_value=[])
         ap.embedding_models_service.get_embedding_models_by_provider = AsyncMock(return_value=[])
-        ap.rerank_models_service.get_rerank_models_by_provider = AsyncMock(
-            return_value=[{'name': 'Qwen3-Reranker-8B'}]
-        )
+        ap.rerank_models_service.get_rerank_models_by_provider = AsyncMock(return_value=[{'name': 'Qwen3-Reranker-8B'}])
 
-        result = await ModelProviderService(ap).scan_provider_models('rerank-scan-uuid', model_type='rerank')
+        result = await ModelProviderService(ap).scan_provider_models(
+            WORKSPACE_UUID, 'rerank-scan-uuid', model_type='rerank'
+        )
 
         assert result['models'][0]['type'] == 'rerank'
         assert result['models'][0]['already_added'] is True

tests/unit_tests/api/service/test_user_service.py

@@ -418,6 +418,63 @@ class TestUserServiceGenerateJwtToken:
         assert token is not None
 
 
+
+    async def test_admin_owner_token_is_scoped_and_capped_at_five_minutes(self):
+        ap = SimpleNamespace()
+        ap.instance_config = SimpleNamespace()
+        ap.instance_config.data = {'system': {'jwt': {'secret': 'test_secret', 'expire': 7200}}}
+        service = UserService(ap)
+        before = datetime.datetime.now(datetime.timezone.utc).timestamp()
+
+        token = await service.generate_jwt_token(
+            'owner@example.com',
+            admin_owner_scope={
+                'actor_account_uuid': 'admin-uuid',
+                'workspace_uuid': 'workspace-uuid',
+                'effective_role': 'owner',
+            },
+        )
+        payload = jwt.decode(
+            token,
+            'test_secret',
+            algorithms=['HS256'],
+            options={'verify_aud': False},
+        )
+
+        assert payload['exp'] <= before + 301
+        assert service.get_admin_owner_scope(token) == {
+            'actor_account_uuid': 'admin-uuid',
+            'workspace_uuid': 'workspace-uuid',
+            'effective_role': 'owner',
+        }
+
+        legacy_token = jwt.encode(
+            {
+                'user': 'legacy@example.com',
+                'exp': datetime.datetime.now(datetime.timezone.utc) + datetime.timedelta(minutes=5),
+            },
+            'test_secret',
+            algorithm='HS256',
+        )
+        assert service.get_admin_owner_scope(legacy_token) is None
+
+    async def test_admin_owner_token_rejects_invalid_scope(self):
+        ap = SimpleNamespace()
+        ap.instance_config = SimpleNamespace()
+        ap.instance_config.data = {'system': {'jwt': {'secret': 'test_secret', 'expire': 7200}}}
+        service = UserService(ap)
+
+        with pytest.raises(ValueError, match='Invalid admin owner token scope'):
+            await service.generate_jwt_token(
+                'owner@example.com',
+                admin_owner_scope={
+                    'actor_account_uuid': 'admin-uuid',
+                    'workspace_uuid': 'workspace-uuid',
+                    'effective_role': 'member',
+                },
+            )
+
+
 class TestUserServiceVerifyJwtToken:
     """Tests for verify_jwt_token method."""
 

tests/unit_tests/cloud/test_space_launch.py

@@ -60,6 +60,7 @@ def _service(private_key: Ed25519PrivateKey, *, now: int) -> SpaceLaunchService:
     app = SimpleNamespace(
         deployment=SimpleNamespace(multi_workspace_enabled=True, verification_key_id=KEY_ID),
         workspace_service=SimpleNamespace(instance_uuid=INSTANCE_UUID),
+        logger=SimpleNamespace(info=lambda *args, **kwargs: None),
         instance_config=SimpleNamespace(
             data={
                 'space': {
@@ -86,6 +87,58 @@ async def test_consumes_valid_workspace_launch_assertion_once():
         await service.consume_assertion(token, expected_workspace_uuid=WORKSPACE_UUID)
 
 
+
+
+async def test_consumes_admin_owner_launch_once_and_validates_claims():
+    private_key = Ed25519PrivateKey.generate()
+    now = int(time.time())
+    service = _service(private_key, now=now)
+    claims = _claims(now=now)
+    claims['payload'].update(
+        {
+            'launch_mode': 'admin_owner',
+            'actor_account_uuid': '33333333-3333-4333-8333-333333333333',
+            'effective_role': 'owner',
+        }
+    )
+    token = _sign(private_key, claims)
+
+    launch = await service.consume_assertion(token, expected_workspace_uuid=WORKSPACE_UUID)
+
+    assert launch == {
+        'account_uuid': ACCOUNT_UUID,
+        'workspace_uuid': WORKSPACE_UUID,
+        'launch_mode': 'admin_owner',
+        'actor_account_uuid': '33333333-3333-4333-8333-333333333333',
+        'effective_role': 'owner',
+    }
+    with pytest.raises(SpaceLaunchError, match='already been consumed'):
+        await service.consume_assertion(token, expected_workspace_uuid=WORKSPACE_UUID)
+
+    invalid = _claims(now=now)
+    invalid['payload'].update(
+        {
+            'launch_mode': 'admin_owner',
+            'actor_account_uuid': '33333333-3333-4333-8333-333333333333',
+            'effective_role': 'member',
+        }
+    )
+    with pytest.raises(SpaceLaunchError, match='effective role'):
+        await service.consume_assertion(_sign(private_key, invalid), expected_workspace_uuid=WORKSPACE_UUID)
+
+    too_long = _claims(now=now)
+    too_long['exp'] = now + 91
+    too_long['payload'].update(
+        {
+            'launch_mode': 'admin_owner',
+            'actor_account_uuid': '33333333-3333-4333-8333-333333333333',
+            'effective_role': 'owner',
+        }
+    )
+    with pytest.raises(SpaceLaunchError, match='lifetime exceeds 90 seconds'):
+        await service.consume_assertion(_sign(private_key, too_long), expected_workspace_uuid=WORKSPACE_UUID)
+
+
 async def test_replay_cache_does_not_scan_all_live_assertions(monkeypatch):
     private_key = Ed25519PrivateKey.generate()
     now = int(time.time())

tests/unit_tests/provider/test_model_manager.py

@@ -114,9 +114,10 @@ async def test_sync_new_models_from_space_creates_rerank_models(mock_app_for_mod
     app.rerank_models_service.get_rerank_models = AsyncMock(return_value=[])
 
     model_mgr = ModelManager(app)
-    await model_mgr.sync_new_models_from_space()
+    await model_mgr.sync_new_models_from_space(TEST_EXECUTION_CONTEXT)
 
     app.rerank_models_service.create_rerank_model.assert_awaited_once_with(
+        TEST_EXECUTION_CONTEXT,
         {
             'uuid': 'rerank-model-uuid',
             'name': 'Qwen3-Reranker-8B',
@@ -1109,4 +1110,4 @@ def test_provider_not_found_error_str():
     error = provider_errors.ProviderNotFoundError('test-provider')
 
     assert str(error) == 'Provider test-provider not found'
-    assert error.provider_name == 'test-provider'
+    assert error.provider_name == 'test-provider'