test(persistence): track current migration head

Co-authored-by: Chan <dadachann@users.noreply.github.com>

.github/workflows/deploy-prod.yml

@@ -15,7 +15,7 @@ concurrency:
 env:
   CORE_IMAGE: ${{ secrets.DOCKER_USERNAME }}/langbot
   CLOUD_IMAGE: ${{ secrets.DOCKER_USERNAME }}/langbot-cloud-core
-  SPACE_REF: e1b261dac45e886efc667b1096a4ec493c6a6111
+  SPACE_REF: 58253c53933f95d81b035fbe2efedb55b6c1a82b
 
 jobs:
   build-and-deploy:
@@ -57,22 +57,3 @@ jobs:
             ${{ env.CLOUD_IMAGE }}:deploy-prod
           cache-from: type=gha,scope=cloud-core-prod
           cache-to: type=gha,mode=max,scope=cloud-core-prod
-      - name: Configure SSH
-        env:
-          SSH_KEY: ${{ secrets.JP09_SSH_KEY }}
-          KNOWN_HOSTS: ${{ secrets.JP09_KNOWN_HOSTS }}
-        run: |
-          install -m 700 -d ~/.ssh
-          install -m 600 /dev/null ~/.ssh/id_ed25519
-          printf '%s\n' "$SSH_KEY" > ~/.ssh/id_ed25519
-          printf '%s\n' "$KNOWN_HOSTS" > ~/.ssh/known_hosts
-      - name: Upload release manifest and deploy
-        env:
-          HOST: ${{ secrets.JP09_HOST }}
-          USER: ${{ secrets.JP09_USER }}
-          PORT: ${{ secrets.JP09_PORT }}
-        run: |
-          remote="$USER@$HOST"
-          ssh -p "$PORT" "$remote" 'install -d -m 700 /opt/langbot-cloud-prod'
-          scp -P "$PORT" deploy/prod/docker-compose.yml deploy/prod/deploy.sh "$remote:/opt/langbot-cloud-prod/"
-          ssh -p "$PORT" "$remote" "chmod 700 /opt/langbot-cloud-prod/deploy.sh && /opt/langbot-cloud-prod/deploy.sh prod-${GITHUB_SHA}"

deploy/prod/deploy.sh

@@ -15,6 +15,10 @@ grep -Fq 'SPACE__URL: https://space.langbot.app' <<<"$rendered_compose" || {
   echo 'Cloud user-facing Space URL must be https://space.langbot.app' >&2
   exit 5
 }
+grep -Eq 'LANGBOT_TELEMETRY_INGEST_TOKEN: .+' <<<"$rendered_compose" || {
+  echo 'Cloud telemetry ingest token must be configured' >&2
+  exit 6
+}
 
 update_env() {
   local key=$1 value=$2
@@ -44,6 +48,7 @@ update_env LANGBOT_IMAGE_TAG "$TAG"
 set -a
 . ./.env
 set +a
+: "${CLOUD_V2_CONTROL_PLANE_TOKEN:?CLOUD_V2_CONTROL_PLANE_TOKEN is required}"
 
 for attempt in 1 2 3 4 5; do
   if docker compose pull postgres redis migrate plugin-runtime core; then

deploy/prod/docker-compose.yml

@@ -88,6 +88,7 @@ services:
       MCP__STDIO__ENABLED: "false"
       LANGBOT_SPACE_CONTROL_PLANE_URL: https://space.langbot.app
       LANGBOT_SPACE_CONTROL_PLANE_TOKEN: ${CLOUD_V2_CONTROL_PLANE_TOKEN}
+      LANGBOT_TELEMETRY_INGEST_TOKEN: ${CLOUD_V2_CONTROL_PLANE_TOKEN}
       LANGBOT_SPACE_CONTROL_PLANE_PUBLIC_KEY: ${CLOUD_V2_MANIFEST_PUBLIC_KEY}
       LANGBOT_SPACE_CONTROL_PLANE_KEY_ID: ${CLOUD_V2_MANIFEST_KEY_ID}
       SPACE__URL: https://space.langbot.app

docker/docker-compose.yaml

@@ -14,8 +14,8 @@ services:
     restart: on-failure
     environment:
       - TZ=Asia/Shanghai
-      # Shared with the langbot service and sent only as a WebSocket handshake
-      # header. Generate with: openssl rand -hex 32
+      # Optional. Leave unset on both OSS services, or set the same value on
+      # both to protect the control WebSocket. Generate with: openssl rand -hex 32
       - LANGBOT_PLUGIN_RUNTIME_CONTROL_TOKEN=${LANGBOT_PLUGIN_RUNTIME_CONTROL_TOKEN:-}
       # Process-wide admission for every asyncio.to_thread() call.
       - LANGBOT_BLOCKING_EXECUTOR_MAX_WORKERS=${LANGBOT_BLOCKING_EXECUTOR_MAX_WORKERS:-8}
@@ -77,8 +77,7 @@ services:
     restart: on-failure
     environment:
       - TZ=Asia/Shanghai
-      # Must match langbot_plugin_runtime. Empty/missing values make the
-      # external control channel fail closed.
+      # Optional. Leave unset on both OSS services, or match plugin Runtime.
       - LANGBOT_PLUGIN_RUNTIME_CONTROL_TOKEN=${LANGBOT_PLUGIN_RUNTIME_CONTROL_TOKEN:-}
       # Must match the value supplied to langbot_box. The token is sent only
       # in WebSocket handshake headers, never in URLs or action payloads.

docs/multi-tenant/implementation-decisions.md

@@ -81,10 +81,10 @@ This log records implementation choices made while delivering the Workspace arch
 - Decision: The MCP ASGI mount authenticates the API key once, binds an immutable per-request `RequestContext`, and every tool checks a fixed permission before calling tenant services with that same context.
 - Reason: Authenticating the transport without propagating Workspace identity into tool calls would leave the direct service path globally scoped.
 
-### Released SDK protocol is pinned from PyPI
+### Unreleased SDK protocol is pinned reproducibly without publishing
 
-- Decision: The SDK tenancy protocol is released as `langbot-plugin==0.5.0` and LangBot pins that exact registry version.
-- Reason: The final PyPI release contains the complete tenant action context and shared Runtime hardening, while the exact version pin keeps production installs reproducible.
+- Decision: The SDK tenancy protocol is versioned as 0.4.18. This task does not create a GitHub release or publish PyPI because the user authorized pushing code, not a package release. After the SDK feature branch is final, LangBot's feature branch temporarily pins the exact pushed SDK Git commit. Before merging to master, the release gate is to publish `langbot-plugin==0.4.18` and replace the Git pin with the registry pin.
+- Reason: The current registry release does not contain the complete tenant action context and shared Runtime hardening. An exact Git commit is reproducible and keeps the feature branch testable without expanding release authority.
 
 ### Cloud directory writes stay outside Core
 

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "langbot"
-version = "4.10.6"
+version = "4.10.7"
 description = "Production-grade platform for building agentic IM bots"
 readme = "README.md"
 license-files = ["LICENSE"]
@@ -71,7 +71,7 @@ dependencies = [
     "chromadb>=1.0.0,<2.0.0",
     "qdrant-client (>=1.15.1,<2.0.0)",
     "pyseekdb==1.1.0.post3",
-    "langbot-plugin==0.5.0",
+    "langbot-plugin @ git+https://github.com/langbot-app/langbot-plugin-sdk.git@9d216208cdfb41f0cb7fcb64632e2a46816d6dc6",
     "asyncpg>=0.30.0",
     "line-bot-sdk>=3.19.0",
     "matrix-nio>=0.25.2",

skills/skills/langbot-deploy/SKILL.md

@@ -32,12 +32,13 @@ The `all` / `box` profile starts three services:
   the LangBot and Box containers. Generate it once with `openssl rand -hex 32`;
   never put it in `box.runtime.endpoint` or commit it to config.
 
-Every Compose deployment also needs one
-`LANGBOT_PLUGIN_RUNTIME_CONTROL_TOKEN` shared by `langbot` and
-`langbot_plugin_runtime`. Generate it with `openssl rand -hex 32` and export it
-before `docker compose up`; the external Plugin Runtime fails closed when the
-token is empty or weak. Kubernetes uses the `langbot-plugin-runtime-control`
-Secret shown in `docker/kubernetes.yaml`.
+A Compose deployment may optionally set
+`LANGBOT_PLUGIN_RUNTIME_CONTROL_TOKEN` on both `langbot` and
+`langbot_plugin_runtime` when port 5400 needs shared-secret protection. OSS
+defaults to leaving it unset on both sides. If enabled, generate one value with
+`openssl rand -hex 32`; configuring only one side causes the control connection
+to fail. Kubernetes may use the `langbot-plugin-runtime-control` Secret shown in
+`docker/kubernetes.yaml`.
 
 With Box off, the dashboard/skills list stays visible (read-only) but sandbox
 tools, skill add/edit, and stdio MCP are disabled. Set `box.enabled: false`

src/langbot/pkg/api/http/authz.py

@@ -19,7 +19,6 @@ class Permission(enum.StrEnum):
     WORKSPACE_VIEW = 'workspace.view'
     WORKSPACE_UPDATE = 'workspace.update'
     WORKSPACE_DELETE = 'workspace.delete'
-    OWNER_TRANSFER = 'owner.transfer'
     MEMBER_VIEW = 'member.view'
     MEMBER_INVITE = 'member.invite'
     MEMBER_UPDATE_ROLE = 'member.update_role'
@@ -49,7 +48,6 @@ _ROLE_PERMISSIONS: typing.Final = types.MappingProxyType(
             if permission
             not in {
                 Permission.WORKSPACE_DELETE,
-                Permission.OWNER_TRANSFER,
                 Permission.BILLING_LINK_MANAGE,
             }
         ),
@@ -74,6 +72,11 @@ class AuthorizationError(Exception):
     error_code = 'forbidden'
 
 
+class AuthenticationDeniedError(AuthorizationError):
+    status_code = 401
+    error_code = 'invalid_authentication'
+
+
 class WorkspaceRequiredError(AuthorizationError):
     status_code = 400
     error_code = 'workspace_required'

src/langbot/pkg/api/http/context.py

@@ -9,6 +9,7 @@ class PrincipalType(enum.StrEnum):
 
     ACCOUNT = 'account'
     API_KEY = 'api_key'
+    SUPPORT_ADMIN = 'support_admin'
     SYSTEM = 'system'
     PUBLIC_BOT = 'public_bot'
 
@@ -19,7 +20,9 @@ class PrincipalContext:
 
     principal_type: PrincipalType
     account_uuid: str | None = None
+    actor_account_uuid: str | None = None
     api_key_uuid: str | None = None
+    support_session_id: str | None = None
 
 
 @dataclasses.dataclass(frozen=True, slots=True)

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

@@ -14,10 +14,18 @@ from ....utils import bounded_executor
 from ....workspace.collaboration import MembershipPermissionError, WorkspaceCollaborationError
 from ....workspace.errors import WorkspaceNotFoundError
 from ....cloud.entitlements import EntitlementUnavailableError
-from ....cloud.quotas import WorkspaceQuotaExceededError
 from ....core.errors import TaskCapacityError
-from ..authz import AuthorizationError, Permission, permissions_for_role, require_permission
+from ..authz import (
+    AuthenticationDeniedError,
+    AuthorizationError,
+    Permission,
+    PermissionDeniedError,
+    WorkspaceRequiredError,
+    permissions_for_role,
+    require_permission,
+)
 from ..context import PrincipalContext, PrincipalType, RequestContext, WorkspaceContext
+from ....cloud.support_admin import SupportAdminSessionError
 
 if typing.TYPE_CHECKING:
     from ....core.app import Application
@@ -52,6 +60,16 @@ class AuthType(enum.Enum):
     USER_TOKEN_OR_API_KEY = 'user-token-or-api-key'
 
 
+_SUPPORT_ADMIN_DENIED_PERMISSIONS = frozenset(
+    {
+        Permission.MEMBER_VIEW.value,
+        Permission.MEMBER_INVITE.value,
+        Permission.MEMBER_UPDATE_ROLE.value,
+        Permission.MEMBER_REMOVE.value,
+    }
+)
+
+
 class RouterGroup(abc.ABC):
     name: str
 
@@ -96,6 +114,10 @@ class RouterGroup(abc.ABC):
                         return self.http_status(401, -1, 'No valid user token provided')
 
                     try:
+                        if self._is_support_admin_token(token):
+                            raise AuthenticationDeniedError(
+                                'Support admin tokens cannot be refreshed or used on account endpoints'
+                            )
                         account, user_email = await self._authenticate_account(token)
                         # Account-token routes deliberately stop before Workspace
                         # selection. They may bootstrap a selector, but cannot
@@ -112,8 +134,13 @@ class RouterGroup(abc.ABC):
                         return self.http_status(401, -1, 'No valid user token provided')
 
                     try:
-                        account, user_email = await self._authenticate_account(token)
-                        request_context = await self._resolve_account_context(account, auth_type)
+                        request_context = await self._authenticate_support_admin(token, auth_type)
+                        if request_context is not None:
+                            self._require_support_admin_route_allowed(rule, f, permission)
+                            user_email = None
+                        else:
+                            account, user_email = await self._authenticate_account(token)
+                            request_context = await self._resolve_account_context(account, auth_type)
                         if permission is not None:
                             if request_context is None:
                                 raise AuthorizationError('Workspace authorization is unavailable')
@@ -142,10 +169,20 @@ class RouterGroup(abc.ABC):
                         return self._auth_error_response(e)
 
                 elif auth_type == AuthType.USER_TOKEN_OR_API_KEY:
+                    token = quart.request.headers.get('Authorization', '').replace('Bearer ', '')
+                    if token and self._is_support_admin_token(token):
+                        try:
+                            request_context = await self._authenticate_support_admin(token, auth_type)
+                            if request_context is None:
+                                raise AuthenticationDeniedError('Invalid support admin token')
+                            self._require_support_admin_route_allowed(rule, f, permission)
+                            if permission is not None:
+                                require_permission(request_context, permission)
+                            self._inject_handler_context(f, kwargs, None, request_context)
+                        except Exception as e:
+                            return self._auth_error_response(e)
                     # Try API key first (check X-API-Key header)
-                    api_key = quart.request.headers.get('X-API-Key', '')
-
-                    if api_key:
+                    elif api_key := quart.request.headers.get('X-API-Key', ''):
                         # API key authentication
                         try:
                             request_context = await self._authenticate_api_key(api_key, auth_type)
@@ -156,8 +193,6 @@ class RouterGroup(abc.ABC):
                             return self._auth_error_response(e)
                     else:
                         # Try user token authentication (Authorization header)
-                        token = quart.request.headers.get('Authorization', '').replace('Bearer ', '')
-
                         if not token:
                             return self.http_status(
                                 401, -1, 'No valid authentication provided (user token or API key required)'
@@ -220,8 +255,6 @@ class RouterGroup(abc.ABC):
                         return self.http_status(403, e.code, str(e))
                     if isinstance(e, WorkspaceCollaborationError):
                         return self.http_status(400, e.code, str(e))
-                    if isinstance(e, WorkspaceQuotaExceededError):
-                        return self.http_status(409, e.error_code, str(e))
                     if isinstance(e, TaskCapacityError):
                         return self.http_status(429, 'task_capacity_exceeded', str(e))
                     if isinstance(
@@ -271,10 +304,83 @@ class RouterGroup(abc.ABC):
             raise ValueError('User not found')
         return account, account.user
 
+    def _is_support_admin_token(self, token: str) -> bool:
+        service = getattr(self.ap, 'support_admin_session_service', None)
+        detector = getattr(service, 'is_support_admin_token', None)
+        return callable(detector) and detector(token) is True
+
+    async def _authenticate_support_admin(
+        self,
+        token: str,
+        auth_type: AuthType,
+        *,
+        workspace_uuid: str | None = None,
+        request_id: str | None = None,
+    ) -> RequestContext | None:
+        service = getattr(self.ap, 'support_admin_session_service', None)
+        detector = getattr(service, 'is_support_admin_token', None)
+        if service is None or not callable(detector) or detector(token) is not True:
+            return None
+
+        requested_workspace_uuid = (
+            workspace_uuid if workspace_uuid is not None else quart.request.headers.get('X-Workspace-Id')
+        )
+        if not requested_workspace_uuid:
+            raise WorkspaceRequiredError('Support admin token requires an explicit Workspace selector')
+        try:
+            identity = await service.authenticate_token(
+                token,
+                requested_workspace_uuid=requested_workspace_uuid,
+            )
+        except SupportAdminSessionError as exc:
+            raise AuthenticationDeniedError(str(exc)) from exc
+
+        entitlement_revision = await self._resolve_entitlement_revision(
+            identity.instance_uuid,
+            identity.workspace_uuid,
+        )
+        request_context = RequestContext(
+            instance_uuid=identity.instance_uuid,
+            placement_generation=identity.placement_generation,
+            request_id=request_id or self.request_id(),
+            auth_type=auth_type.value,
+            principal=PrincipalContext(
+                principal_type=PrincipalType.SUPPORT_ADMIN,
+                actor_account_uuid=identity.actor_account_uuid,
+                support_session_id=identity.grant_jti_hash,
+            ),
+            workspace=WorkspaceContext(
+                workspace_uuid=identity.workspace_uuid,
+                membership_uuid=None,
+                role='owner',
+                permissions=permissions_for_role('owner') - _SUPPORT_ADMIN_DENIED_PERMISSIONS,
+                membership_revision=0,
+            ),
+            entitlement_revision=entitlement_revision,
+        )
+        quart.g.request_context = request_context
+        quart.g.workspace_membership = None
+        return request_context
+
+    @staticmethod
+    def _require_support_admin_route_allowed(
+        rule: str,
+        handler: RouteCallable,
+        permission: Permission | str | None,
+    ) -> None:
+        parameters = inspect.signature(handler).parameters
+        if rule.startswith('/api/v1/user/') or 'account' in parameters or 'user_email' in parameters:
+            raise AuthenticationDeniedError('Support admin tokens are not permitted on account endpoints')
+        permission_value = permission.value if isinstance(permission, Permission) else permission
+        if permission_value in _SUPPORT_ADMIN_DENIED_PERMISSIONS:
+            raise PermissionDeniedError(permission_value)
+
     async def _resolve_account_context(
         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)

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

@@ -97,6 +97,16 @@ class WebSocketChatRouterGroup(group.RouterGroup):
         if not token or not workspace_uuid:
             raise ValueError('Authentication is required')
 
+        support_context = await self._authenticate_support_admin(
+            token,
+            group.AuthType.USER_TOKEN,
+            workspace_uuid=workspace_uuid,
+            request_id=quart.websocket.headers.get('X-Request-Id') or str(uuid.uuid4()),
+        )
+        if support_context is not None:
+            require_permission(support_context, Permission.RUNTIME_OPERATE)
+            return support_context, token
+
         account, _ = await self._authenticate_account(token)
         account_uuid = getattr(account, 'uuid', None)
         collaboration_service = getattr(self.ap, 'workspace_collaboration_service', None)
@@ -131,6 +141,23 @@ class WebSocketChatRouterGroup(group.RouterGroup):
     ) -> RequestContext:
         """Recheck revocable account, membership, permission, and placement state."""
 
+        if request_context.principal.principal_type == PrincipalType.SUPPORT_ADMIN:
+            current_context = await self._authenticate_support_admin(
+                token,
+                group.AuthType.USER_TOKEN,
+                workspace_uuid=request_context.workspace_uuid,
+                request_id=request_context.request_id,
+            )
+            if current_context is None or current_context.principal != request_context.principal:
+                raise ValueError('WebSocket support admin session changed')
+            if (
+                current_context.instance_uuid != request_context.instance_uuid
+                or current_context.placement_generation != request_context.placement_generation
+            ):
+                raise ValueError('WebSocket authorization changed')
+            require_permission(current_context, Permission.RUNTIME_OPERATE)
+            return current_context
+
         account, _ = await self._authenticate_account(token)
         account_uuid = getattr(account, 'uuid', None)
         if account_uuid != request_context.account_uuid:
@@ -211,6 +238,7 @@ class WebSocketChatRouterGroup(group.RouterGroup):
                     scope=WebSocketScope.from_context(request_context),
                     pipeline_uuid=pipeline_uuid,
                     session_type=session_type,
+                    trigger_principal=request_context.principal,
                     metadata={'user_agent': quart.websocket.headers.get('User-Agent', '')},
                     send_queue_size=(
                         self.ap.instance_config.data.get('system', {})
@@ -391,7 +419,7 @@ class WebSocketChatRouterGroup(group.RouterGroup):
                         )
                     elif message_type == 'message':
                         try:
-                            await self._revalidate_websocket_authorization(request_context, token)
+                            request_context = await self._revalidate_websocket_authorization(request_context, token)
                         except Exception:
                             await connection.send_queue.put({'type': 'error', 'message': 'Unauthorized'})
                             break

src/langbot/pkg/api/http/controller/groups/platform/adapters.py

@@ -22,6 +22,7 @@ class _AdapterSessionScope:
     principal_type: str
     account_uuid: str | None
     api_key_uuid: str | None
+    support_session_id: str | None
 
     @classmethod
     def from_request_context(cls, request_context: RequestContext) -> '_AdapterSessionScope':
@@ -33,6 +34,7 @@ class _AdapterSessionScope:
             principal_type=principal.principal_type.value,
             account_uuid=principal.account_uuid,
             api_key_uuid=principal.api_key_uuid,
+            support_session_id=principal.support_session_id,
         )
 
     def matches(self, request_context: RequestContext) -> bool:

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

@@ -392,8 +392,8 @@ class PluginsRouterGroup(group.RouterGroup):
         )
         async def _(request_context: RequestContext) -> str:
             """Get plugin debug information including debug URL and key"""
-            await self._require_authenticated_plugin_runtime_context(request_context)
-            debug_info = await self.ap.plugin_connector.get_debug_info()
+            execution_context = await self._require_authenticated_plugin_runtime_context(request_context)
+            debug_info = await self.ap.plugin_connector.get_debug_info(execution_context)
 
             # Get debug URL from config
             plugin_config = self.ap.instance_config.data.get('plugin', {})
@@ -403,6 +403,7 @@ class PluginsRouterGroup(group.RouterGroup):
                 data={
                     'debug_url': debug_url,
                     'plugin_debug_key': debug_info.get('plugin_debug_key', ''),
+                    'expires_at': debug_info.get('expires_at', ''),
                 }
             )
 

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

@@ -1,6 +1,7 @@
 import quart
 import argon2
 import asyncio
+import datetime
 import uuid
 from urllib.parse import parse_qs, urlsplit
 
@@ -218,7 +219,22 @@ class UserRouterGroup(group.RouterGroup):
             try:
                 consumed_state = await self.ap.user_service.consume_space_oauth_state_details(state, 'login')
                 # Exchange code for tokens
-                token_data = await self.ap.space_service.exchange_oauth_code(code)
+                launch_workspace_uuid = consumed_state.launch_workspace_uuid
+                workspace_uuids = [launch_workspace_uuid] if launch_workspace_uuid else []
+                workspace_created_ats: dict[str, int] = {}
+                if not workspace_uuids and getattr(getattr(self.ap, 'deployment', None), 'mode', 'oss') != 'cloud':
+                    binding = await self.ap.workspace_service.get_execution_binding()
+                    workspace_uuids = [binding.workspace_uuid]
+                    workspace_created_at = binding.workspace_created_at
+                    if workspace_created_at is not None:
+                        if workspace_created_at.tzinfo is None:
+                            workspace_created_at = workspace_created_at.replace(tzinfo=datetime.UTC)
+                        workspace_created_ats[binding.workspace_uuid] = int(workspace_created_at.timestamp())
+                token_data = await self.ap.space_service.exchange_oauth_code(
+                    code,
+                    workspace_uuids,
+                    workspace_created_ats,
+                )
                 access_token = token_data.get('access_token')
                 refresh_token = token_data.get('refresh_token')
                 expires_in = token_data.get('expires_in', 0)
@@ -231,7 +247,6 @@ class UserRouterGroup(group.RouterGroup):
                     access_token, refresh_token, expires_in
                 )
 
-                launch_workspace_uuid = consumed_state.launch_workspace_uuid
                 if launch_workspace_uuid:
                     try:
                         access = await self.ap.workspace_collaboration_service.resolve_account_workspace(
@@ -285,8 +300,25 @@ class UserRouterGroup(group.RouterGroup):
                 request_context.workspace_uuid,
             )
             owner = await self.ap.user_service.get_workspace_owner(access.workspace.uuid)
-            owner_space_bound = bool(owner and owner.space_account_uuid)
-            credits = await self.ap.space_service.get_credits(owner.user) if owner_space_bound else None
+            cloud_mode = getattr(getattr(self.ap, 'deployment', None), 'mode', 'oss') == 'cloud'
+            owner_has_local_space_credentials = bool(owner and owner.space_account_uuid)
+            # Cloud Accounts authenticate through LangBot Account, so every projected
+            # Workspace owner is already bound even when this Core has no local OAuth
+            # token row (model billing uses the owner's control-plane API key).
+            owner_space_bound = cloud_mode or owner_has_local_space_credentials
+            if cloud_mode:
+                catalog_service = getattr(self.ap, 'cloud_model_catalog_service', None)
+                credits = (
+                    catalog_service.get_workspace_credits(access.workspace.uuid)
+                    if catalog_service is not None
+                    else None
+                )
+            else:
+                credits = (
+                    await self.ap.space_service.get_credits(owner.user)
+                    if owner is not None and owner.space_account_uuid
+                    else None
+                )
             return self.success(
                 data={
                     'credits': credits,
@@ -302,8 +334,10 @@ class UserRouterGroup(group.RouterGroup):
                 return self.success(data={'initialized': False})
 
             capabilities = await self.ap.user_service.get_login_capabilities()
-            if getattr(getattr(self.ap, 'deployment', None), 'mode', 'oss') == 'cloud':
+            cloud_mode = getattr(getattr(self.ap, 'deployment', None), 'mode', 'oss') == 'cloud'
+            if cloud_mode:
                 capabilities['password_login_enabled'] = False
+            capabilities['authenticated_invitation_acceptance_enabled'] = cloud_mode
             return self.success(data={'initialized': True, **capabilities})
 
         @self.route('/set-password', methods=['POST'], auth_type=group.AuthType.USER_TOKEN)
@@ -396,6 +430,19 @@ class UserRouterGroup(group.RouterGroup):
                 launch_assertion,
                 expected_workspace_uuid=workspace_uuid,
             )
+            if launch.get('launch_mode') == 'support_admin':
+                token = launch.get('support_admin_token')
+                if not token:
+                    raise SpaceLaunchError('Support admin launch session was not issued')
+                return self.success(
+                    data={
+                        'token': token,
+                        'workspace_uuid': launch['workspace_uuid'],
+                        'principal_type': 'support_admin',
+                        'actor_account_uuid': launch['actor_account_uuid'],
+                    }
+                )
+
             account = await self.ap.user_service.get_user_by_uuid(launch['account_uuid'])
             if account is None:
                 raise SpaceLaunchError('Launch Account is not projected into Core')
@@ -410,7 +457,6 @@ class UserRouterGroup(group.RouterGroup):
                     'token': token,
                     'user': account.user,
                     'workspace_uuid': access.workspace.uuid,
-                    'return_path': launch.get('return_path', '/home'),
                 }
             )
         except SpaceLaunchError:

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

@@ -5,7 +5,7 @@ import typing
 import quart
 
 from ...authz import Permission, permissions_for_role
-from ...context import RequestContext
+from ...context import PrincipalType, RequestContext
 from ...service.user import AccountExistsLoginRequiredError, ControlPlaneDirectoryRequiredError
 from .....entity.persistence.workspace import Workspace, WorkspaceInvitation, WorkspaceMembership
 from .....entity.persistence.workspace import WorkspaceSource
@@ -30,12 +30,14 @@ def _workspace_payload(workspace: Workspace) -> dict[str, typing.Any]:
 def _membership_payload(
     membership: WorkspaceMembership,
     *,
+    display_name: str,
     email: str,
 ) -> dict[str, typing.Any]:
     return {
         'uuid': membership.uuid,
         'workspace_uuid': membership.workspace_uuid,
         'account_uuid': membership.account_uuid,
+        'display_name': display_name,
         'email': email,
         'role': membership.role,
         'status': membership.status,
@@ -94,7 +96,11 @@ class WorkspacesRouterGroup(group.RouterGroup):
                 workspaces.append(
                     {
                         'workspace': _workspace_payload(access.workspace),
-                        'membership': _membership_payload(access.membership, email=account.user),
+                        'membership': _membership_payload(
+                            access.membership,
+                            display_name=account.user,
+                            email=account.normalized_email,
+                        ),
                         'permissions': sorted(permissions_for_role(access.membership.role)),
                         'placement_generation': access.execution.placement_generation,
                         'plan_name': plan_name,
@@ -120,9 +126,6 @@ class WorkspacesRouterGroup(group.RouterGroup):
         @self.route('/current', methods=['GET'], permission=Permission.WORKSPACE_VIEW)
         async def _(request_context: RequestContext) -> typing.Any:
             membership = quart.g.workspace_membership
-            account = await self.ap.user_service.get_user_by_uuid(request_context.account_uuid)
-            if account is None:
-                return self.http_status(401, 'invalid_authentication', 'Account not found')
             workspace = await self.ap.workspace_service.get_workspace(request_context.workspace_uuid)
             plan_name: str | None = None
             resolver = getattr(self.ap, 'entitlement_resolver', None)
@@ -132,10 +135,37 @@ class WorkspacesRouterGroup(group.RouterGroup):
                     minimum_revision=request_context.entitlement_revision,
                 )
                 plan_name = entitlement.plan_name
+            if request_context.principal.principal_type == PrincipalType.SUPPORT_ADMIN:
+                return self.success(
+                    data={
+                        'workspace': _workspace_payload(workspace),
+                        'membership': {
+                            'uuid': None,
+                            'workspace_uuid': request_context.workspace_uuid,
+                            'account_uuid': None,
+                            'display_name': None,
+                            'email': None,
+                            'role': 'owner',
+                            'status': 'active',
+                            'joined_at': None,
+                            'created_at': None,
+                        },
+                        'permissions': sorted(request_context.workspace.permissions),
+                        'placement_generation': request_context.placement_generation,
+                        'plan_name': plan_name,
+                    }
+                )
+            account = await self.ap.user_service.get_user_by_uuid(request_context.account_uuid)
+            if account is None:
+                return self.http_status(401, 'invalid_authentication', 'Account not found')
             return self.success(
                 data={
                     'workspace': _workspace_payload(workspace),
-                    'membership': _membership_payload(membership, email=account.user),
+                    'membership': _membership_payload(
+                        membership,
+                        display_name=account.user,
+                        email=account.normalized_email,
+                    ),
                     'permissions': sorted(request_context.workspace.permissions),
                     'placement_generation': request_context.placement_generation,
                     'plan_name': plan_name,
@@ -264,7 +294,8 @@ class WorkspacesRouterGroup(group.RouterGroup):
                 data={
                     'member': _membership_payload(
                         member,
-                        email=account.user if account is not None else '',
+                        display_name=account.user if account is not None else '',
+                        email=account.normalized_email if account is not None else '',
                     )
                 }
             )
@@ -283,7 +314,11 @@ class WorkspacesRouterGroup(group.RouterGroup):
 
     @staticmethod
     def _member_view_payload(view: WorkspaceMemberView) -> dict[str, typing.Any]:
-        return _membership_payload(view.membership, email=view.email)
+        return _membership_payload(
+            view.membership,
+            display_name=view.display_name,
+            email=view.email,
+        )
 
 
 @group.group_class('invitations', '/api/v1/invitations')

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

@@ -4,7 +4,6 @@ import uuid
 import sqlalchemy
 
 from ....core import app
-from ....cloud.quotas import require_resource_capacity, resolve_workspace_quota
 from ....entity.persistence import bot as persistence_bot
 from ....entity.persistence import pipeline as persistence_pipeline
 from ....workspace.errors import WorkspaceNotFoundError
@@ -102,21 +101,20 @@ class BotService:
     async def create_bot(self, context: TenantContext, bot_data: dict) -> str:
         """Create bot"""
         workspace_uuid = require_workspace_uuid(context)
+        # Check limitation
         limitation = self.ap.instance_config.data.get('system', {}).get('limitation', {})
-        quota = await resolve_workspace_quota(
-            self.ap,
-            workspace_uuid,
-            'bots.max',
-            fallback=limitation.get('max_bots', -1),
-        )
+        max_bots = limitation.get('max_bots', -1)
+        if max_bots >= 0:
+            existing_bots = await self.get_bots(context)
+            if len(existing_bots) >= max_bots:
+                raise ValueError(f'Maximum number of bots ({max_bots}) reached')
 
         # TODO: 检查配置信息格式
         bot_data = bot_data.copy()
         bot_data['uuid'] = str(uuid.uuid4())
         bot_data['workspace_uuid'] = workspace_uuid
 
-        # Preserve the legacy flat-row result shape for this optional lookup;
-        # quota admission and insertion below still share one transaction.
+        # bind the most recently updated pipeline if any exist
         result = await self.ap.persistence_mgr.execute_async(
             scope_statement(
                 sqlalchemy.select(persistence_pipeline.LegacyPipeline),
@@ -131,25 +129,7 @@ class BotService:
             bot_data['use_pipeline_uuid'] = pipeline.uuid
             bot_data['use_pipeline_name'] = pipeline.name
 
-        async def persist(execute) -> None:
-            await require_resource_capacity(
-                execute,
-                workspace_uuid=workspace_uuid,
-                model=persistence_bot.Bot,
-                quota=quota,
-                resource_name='bots',
-            )
-
-            await execute(sqlalchemy.insert(persistence_bot.Bot).values(bot_data))
-
-        tenant_uow = getattr(self.ap.persistence_mgr, 'tenant_uow', None)
-        if quota.requires_transaction_lock:
-            if not callable(tenant_uow):
-                raise RuntimeError('Cloud bot quota enforcement requires transactional persistence')
-            async with tenant_uow(workspace_uuid) as uow:
-                await persist(uow.execute)
-        else:
-            await persist(self.ap.persistence_mgr.execute_async)
+        await self.ap.persistence_mgr.execute_async(sqlalchemy.insert(persistence_bot.Bot).values(bot_data))
 
         bot = await self.get_bot(context, bot_data['uuid'], include_secret=True)
 

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

@@ -5,6 +5,7 @@ import uuid
 import sqlalchemy
 from langbot_plugin.api.entities.builtin.provider import message as provider_message
 
+from ....cloud.model_catalog import LANGBOT_MODELS_PROVIDER_REQUESTER
 from ....core import app
 from ....entity.persistence import model as persistence_model
 from ....entity.persistence import pipeline as persistence_pipeline
@@ -113,6 +114,23 @@ async def _require_workspace_provider(
     return provider
 
 
+def _is_cloud_runtime(ap: app.Application) -> bool:
+    mode = getattr(ap.persistence_mgr, 'mode', None)
+    return getattr(mode, 'value', None) == 'cloud_runtime'
+
+
+async def _assert_cloud_managed_provider_mutable(
+    ap: app.Application,
+    context: TenantContext,
+    provider_uuid: str,
+) -> None:
+    if not _is_cloud_runtime(ap):
+        return
+    provider = await _require_workspace_provider(ap, context, provider_uuid)
+    if provider.get('requester') == LANGBOT_MODELS_PROVIDER_REQUESTER:
+        raise ValueError('LangBot Models is managed by Cloud and cannot be modified')
+
+
 async def _require_runtime_provider(
     ap: app.Application,
     context: TenantContext,
@@ -213,6 +231,7 @@ class LLMModelsService:
                 model_data['provider_uuid'] = provider_uuid
 
         await _require_workspace_provider(self.ap, context, model_data['provider_uuid'])
+        await _assert_cloud_managed_provider_mutable(self.ap, context, model_data['provider_uuid'])
         await _validate_provider_supports(self.ap, context, model_data['provider_uuid'], 'llm')
 
         await self.ap.persistence_mgr.execute_async(sqlalchemy.insert(persistence_model.LLMModel).values(**model_data))
@@ -291,11 +310,17 @@ class LLMModelsService:
 
         return model_dict
 
-    async def update_llm_model(self, context: TenantContext, model_uuid: str, model_data: dict) -> None:
+    async def update_llm_model(
+        self,
+        context: TenantContext,
+        model_uuid: str,
+        model_data: dict,
+    ) -> None:
         """Update an existing LLM model"""
         existing_model = await self.get_llm_model(context, model_uuid, include_secret=True)
         if existing_model is None:
             raise WorkspaceNotFoundError('Model not found')
+        await _assert_cloud_managed_provider_mutable(self.ap, context, existing_model['provider_uuid'])
         model_data = model_data.copy()
         model_data.pop('uuid', None)
         model_data.pop('workspace_uuid', None)
@@ -321,6 +346,7 @@ class LLMModelsService:
 
         provider_uuid = model_data.get('provider_uuid', existing_model['provider_uuid'])
         await _require_workspace_provider(self.ap, context, provider_uuid)
+        await _assert_cloud_managed_provider_mutable(self.ap, context, provider_uuid)
         await _validate_provider_supports(self.ap, context, provider_uuid, 'llm')
 
         result = await self.ap.persistence_mgr.execute_async(
@@ -355,6 +381,11 @@ class LLMModelsService:
 
     async def delete_llm_model(self, context: TenantContext, model_uuid: str) -> None:
         """Delete an LLM model"""
+        if _is_cloud_runtime(self.ap):
+            existing_model = await self.get_llm_model(context, model_uuid, include_secret=True)
+            if existing_model is None:
+                raise WorkspaceNotFoundError('Model not found')
+            await _assert_cloud_managed_provider_mutable(self.ap, context, existing_model['provider_uuid'])
         result = await self.ap.persistence_mgr.execute_async(
             scope_statement(
                 sqlalchemy.delete(persistence_model.LLMModel).where(persistence_model.LLMModel.uuid == model_uuid),
@@ -448,7 +479,10 @@ class EmbeddingModelsService:
         return serialized if include_secret else [_redact_model_secrets(model) for model in serialized]
 
     async def create_embedding_model(
-        self, context: TenantContext, model_data: dict, preserve_uuid: bool = False
+        self,
+        context: TenantContext,
+        model_data: dict,
+        preserve_uuid: bool = False,
     ) -> str:
         """Create a new embedding model"""
         model_data = model_data.copy()
@@ -472,6 +506,7 @@ class EmbeddingModelsService:
                 model_data['provider_uuid'] = provider_uuid
 
         await _require_workspace_provider(self.ap, context, model_data['provider_uuid'])
+        await _assert_cloud_managed_provider_mutable(self.ap, context, model_data['provider_uuid'])
         await _validate_provider_supports(self.ap, context, model_data['provider_uuid'], 'text-embedding')
 
         await self.ap.persistence_mgr.execute_async(
@@ -530,11 +565,17 @@ class EmbeddingModelsService:
 
         return model_dict
 
-    async def update_embedding_model(self, context: TenantContext, model_uuid: str, model_data: dict) -> None:
+    async def update_embedding_model(
+        self,
+        context: TenantContext,
+        model_uuid: str,
+        model_data: dict,
+    ) -> None:
         """Update an existing embedding model"""
         existing_model = await self.get_embedding_model(context, model_uuid, include_secret=True)
         if existing_model is None:
             raise WorkspaceNotFoundError('Model not found')
+        await _assert_cloud_managed_provider_mutable(self.ap, context, existing_model['provider_uuid'])
         model_data = model_data.copy()
         model_data.pop('uuid', None)
         model_data.pop('workspace_uuid', None)
@@ -559,6 +600,7 @@ class EmbeddingModelsService:
 
         provider_uuid = model_data.get('provider_uuid', existing_model['provider_uuid'])
         await _require_workspace_provider(self.ap, context, provider_uuid)
+        await _assert_cloud_managed_provider_mutable(self.ap, context, provider_uuid)
         await _validate_provider_supports(self.ap, context, provider_uuid, 'text-embedding')
 
         result = await self.ap.persistence_mgr.execute_async(
@@ -593,6 +635,11 @@ class EmbeddingModelsService:
 
     async def delete_embedding_model(self, context: TenantContext, model_uuid: str) -> None:
         """Delete an embedding model"""
+        if _is_cloud_runtime(self.ap):
+            existing_model = await self.get_embedding_model(context, model_uuid, include_secret=True)
+            if existing_model is None:
+                raise WorkspaceNotFoundError('Model not found')
+            await _assert_cloud_managed_provider_mutable(self.ap, context, existing_model['provider_uuid'])
         result = await self.ap.persistence_mgr.execute_async(
             scope_statement(
                 sqlalchemy.delete(persistence_model.EmbeddingModel).where(
@@ -685,7 +732,12 @@ class RerankModelsService:
         serialized = [self.ap.persistence_mgr.serialize_model(persistence_model.RerankModel, m) for m in models]
         return serialized if include_secret else [_redact_model_secrets(model) for model in serialized]
 
-    async def create_rerank_model(self, context: TenantContext, model_data: dict, preserve_uuid: bool = False) -> str:
+    async def create_rerank_model(
+        self,
+        context: TenantContext,
+        model_data: dict,
+        preserve_uuid: bool = False,
+    ) -> str:
         """Create a new rerank model"""
         model_data = model_data.copy()
         if not preserve_uuid:
@@ -708,6 +760,7 @@ class RerankModelsService:
                 model_data['provider_uuid'] = provider_uuid
 
         await _require_workspace_provider(self.ap, context, model_data['provider_uuid'])
+        await _assert_cloud_managed_provider_mutable(self.ap, context, model_data['provider_uuid'])
         await _validate_provider_supports(self.ap, context, model_data['provider_uuid'], 'rerank')
 
         await self.ap.persistence_mgr.execute_async(
@@ -766,11 +819,17 @@ class RerankModelsService:
 
         return model_dict
 
-    async def update_rerank_model(self, context: TenantContext, model_uuid: str, model_data: dict) -> None:
+    async def update_rerank_model(
+        self,
+        context: TenantContext,
+        model_uuid: str,
+        model_data: dict,
+    ) -> None:
         """Update an existing rerank model"""
         existing_model = await self.get_rerank_model(context, model_uuid, include_secret=True)
         if existing_model is None:
             raise WorkspaceNotFoundError('Model not found')
+        await _assert_cloud_managed_provider_mutable(self.ap, context, existing_model['provider_uuid'])
         model_data = model_data.copy()
         model_data.pop('uuid', None)
         model_data.pop('workspace_uuid', None)
@@ -795,6 +854,7 @@ class RerankModelsService:
 
         provider_uuid = model_data.get('provider_uuid', existing_model['provider_uuid'])
         await _require_workspace_provider(self.ap, context, provider_uuid)
+        await _assert_cloud_managed_provider_mutable(self.ap, context, provider_uuid)
         await _validate_provider_supports(self.ap, context, provider_uuid, 'rerank')
 
         result = await self.ap.persistence_mgr.execute_async(
@@ -829,6 +889,11 @@ class RerankModelsService:
 
     async def delete_rerank_model(self, context: TenantContext, model_uuid: str) -> None:
         """Delete a rerank model"""
+        if _is_cloud_runtime(self.ap):
+            existing_model = await self.get_rerank_model(context, model_uuid, include_secret=True)
+            if existing_model is None:
+                raise WorkspaceNotFoundError('Model not found')
+            await _assert_cloud_managed_provider_mutable(self.ap, context, existing_model['provider_uuid'])
         result = await self.ap.persistence_mgr.execute_async(
             scope_statement(
                 sqlalchemy.delete(persistence_model.RerankModel).where(

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

@@ -5,6 +5,7 @@ import traceback
 
 import sqlalchemy
 
+from ....cloud.model_catalog import LANGBOT_MODELS_PROVIDER_REQUESTER
 from ....core import app
 from ....entity.persistence import model as persistence_model
 from ....workspace.errors import WorkspaceNotFoundError
@@ -20,6 +21,20 @@ class ModelProviderService:
     def __init__(self, ap: app.Application) -> None:
         self.ap = ap
 
+    def _is_cloud_runtime(self) -> bool:
+        mode = getattr(self.ap.persistence_mgr, 'mode', None)
+        return getattr(mode, 'value', None) == 'cloud_runtime'
+
+    def _system_requester_is_reserved(self, requester: object) -> bool:
+        return self._is_cloud_runtime() and requester == LANGBOT_MODELS_PROVIDER_REQUESTER
+
+    async def _assert_provider_mutable(self, context: TenantContext, provider_uuid: str) -> None:
+        if not self._is_cloud_runtime():
+            return
+        provider = await self.get_provider(context, provider_uuid)
+        if provider is not None and self._system_requester_is_reserved(provider.get('requester')):
+            raise ValueError('LangBot Models is managed by Cloud and cannot be modified')
+
     @staticmethod
     def _normalize_api_keys(api_keys: str | list[str] | tuple[str, ...] | None) -> list[str]:
         if api_keys is None:
@@ -99,6 +114,8 @@ class ModelProviderService:
     async def create_provider(self, context: TenantContext, provider_data: dict) -> str:
         """Create a new provider"""
         provider_data = provider_data.copy()
+        if self._system_requester_is_reserved(provider_data.get('requester')):
+            raise ValueError('space-chat-completions is reserved for the Cloud-managed LangBot Models provider')
         provider_data['uuid'] = str(uuid.uuid4())
         provider_data['workspace_uuid'] = require_workspace_uuid(context)
         provider_data['api_keys'] = self._normalize_api_keys(
@@ -115,7 +132,10 @@ class ModelProviderService:
 
     async def update_provider(self, context: TenantContext, provider_uuid: str, provider_data: dict) -> None:
         """Update an existing provider"""
+        await self._assert_provider_mutable(context, provider_uuid)
         provider_data = provider_data.copy()
+        if self._system_requester_is_reserved(provider_data.get('requester')):
+            raise ValueError('space-chat-completions is reserved for the Cloud-managed LangBot Models provider')
         provider_data.pop('uuid', None)
         provider_data.pop('workspace_uuid', None)
         if 'api_keys' in provider_data:
@@ -145,6 +165,7 @@ class ModelProviderService:
 
     async def delete_provider(self, context: TenantContext, provider_uuid: str) -> None:
         """Delete a provider (only if no models reference it)"""
+        await self._assert_provider_mutable(context, provider_uuid)
         workspace_uuid = require_workspace_uuid(context)
         # Check if any models use this provider
         llm_result = await self.ap.persistence_mgr.execute_async(
@@ -245,6 +266,8 @@ class ModelProviderService:
         api_keys: list,
     ) -> str:
         """Find existing provider or create new one"""
+        if self._system_requester_is_reserved(requester):
+            raise ValueError('space-chat-completions is reserved for the Cloud-managed LangBot Models provider')
         workspace_uuid = require_workspace_uuid(context)
         api_keys = self._normalize_api_keys(restore_secret_placeholders(api_keys, sensitive=True))
 

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

@@ -59,6 +59,10 @@ class SpaceService:
         result_list = result.all()
         return result_list[0] if result_list else None
 
+    async def get_valid_access_token(self, user_email: str) -> str | None:
+        """Return a current Space bearer, refreshing and persisting it when needed."""
+        return await self._ensure_valid_token(user_email)
+
     async def _ensure_valid_token(self, user_email: str) -> str | None:
         """Ensure access token is valid, refresh if expired. Returns valid access_token or None."""
         user_obj = await self._get_user_by_email(user_email)
@@ -117,7 +121,12 @@ class SpaceService:
             params['state'] = state
         return f'{authorize_url}?{urlencode(params)}'
 
-    async def exchange_oauth_code(self, code: str) -> typing.Dict:
+    async def exchange_oauth_code(
+        self,
+        code: str,
+        workspace_uuids: list[str] | None = None,
+        workspace_created_ats: dict[str, int] | None = None,
+    ) -> typing.Dict:
         """Exchange OAuth authorization code for tokens"""
         from langbot.pkg.utils import constants
 
@@ -127,7 +136,14 @@ class SpaceService:
         session = httpclient.get_session()
         async with session.post(
             f'{space_url}/api/v1/accounts/oauth/token',
-            json={'code': code, 'instance_id': constants.instance_id},
+            json={
+                'code': code,
+                'instance_id': constants.instance_id,
+                # Sending an explicit empty list tells new Space servers not to
+                # synthesize a legacy instance-derived Workspace binding.
+                'workspace_uuids': workspace_uuids if workspace_uuids is not None else [],
+                'workspace_created_ats': workspace_created_ats or {},
+            },
         ) as response:
             if response.status != 200:
                 error = await httpclient.read_text_limited(response)

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

@@ -400,7 +400,10 @@ 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,
+    ) -> str:
         jwt_secret = self.ap.instance_config.data['system']['jwt']['secret']
         jwt_expire = self.ap.instance_config.data['system']['jwt']['expire']
 
@@ -413,7 +416,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],
@@ -776,8 +779,27 @@ class UserService:
         local_account = await self.get_user_by_email(user_email)
         if local_account is None:
             raise ValueError('User not found')
-        # Exchange code for tokens
-        token_data = await self.ap.space_service.exchange_oauth_code(code)
+        # Exchange code for tokens and bind both installation and the active
+        # OSS Workspace as independent identities.
+        workspace_service = getattr(self.ap, 'workspace_service', None)
+        if workspace_service is not None:
+            binding = await workspace_service.get_execution_binding()
+            created_at = binding.workspace_created_at
+            created_ts = (
+                int(created_at.replace(tzinfo=datetime.timezone.utc).timestamp())
+                if created_at.tzinfo is None
+                else int(created_at.timestamp())
+            )
+            token_data = await self.ap.space_service.exchange_oauth_code(
+                code,
+                [binding.workspace_uuid],
+                {binding.workspace_uuid: created_ts},
+            )
+        else:
+            # Compatibility for early/bootstrap call sites that have not wired
+            # WorkspaceService yet; old Space servers still derive the legacy
+            # Workspace identity from instance_id when the field is omitted.
+            token_data = await self.ap.space_service.exchange_oauth_code(code)
         access_token = token_data.get('access_token')
         refresh_token = token_data.get('refresh_token')
         expires_in = token_data.get('expires_in', 0)

src/langbot/pkg/cloud/bootstrap.py

@@ -13,11 +13,12 @@ from typing import Any, Protocol, runtime_checkable
 from ..workspace.policy import CloudWorkspacePolicy, SingleWorkspacePolicy
 from .directory import DirectoryProjectionProvider, directory_projection_limits_from_config
 from .entitlements import EntitlementProvider, OpenSourceEntitlementProvider
+from .model_catalog import CloudModelCatalogProvider
 
 
 CLOUD_BOOTSTRAP_ENTRY_POINT = 'langbot.cloud_bootstrap'
 REQUIRED_TENANT_ISOLATION_VERSION = 2
-SUPPORTED_PGVECTOR_DIMENSIONS = frozenset({384, 512, 768, 1024, 1536})
+SUPPORTED_PGVECTOR_DIMENSIONS = frozenset({384, 512, 768, 1024, 1536, 3072})
 
 
 class CloudBootstrapError(RuntimeError):
@@ -50,6 +51,7 @@ class OpenSourceDeployment:
     )
     directory_provider: None = None
     manifest_provider: None = None
+    model_catalog_provider: None = None
     persistence_mode: str = 'oss_compat'
     required_vector_backend: str | None = None
 
@@ -80,6 +82,7 @@ class VerifiedCloudDeployment:
     entitlement_provider: EntitlementProvider
     directory_provider: DirectoryProjectionProvider
     manifest_provider: CloudManifestProvider
+    model_catalog_provider: CloudModelCatalogProvider
     verification_key_id: str
     mode: str = dataclasses.field(default='cloud', init=False)
     workspace_policy: CloudWorkspacePolicy = dataclasses.field(default_factory=CloudWorkspacePolicy, init=False)
@@ -110,6 +113,8 @@ class VerifiedCloudDeployment:
             raise CloudBootstrapError('Verified Cloud bootstrap did not provide a directory adapter')
         if not isinstance(self.manifest_provider, CloudManifestProvider):
             raise CloudBootstrapError('Verified Cloud bootstrap did not provide a Manifest renewal adapter')
+        if not isinstance(self.model_catalog_provider, CloudModelCatalogProvider):
+            raise CloudBootstrapError('Verified Cloud bootstrap did not provide a model catalog adapter')
 
     def validate_instance_config(self, config: dict[str, Any]) -> None:
         try:

src/langbot/pkg/cloud/directory_projection.py

@@ -15,6 +15,7 @@ from ..entity.persistence.cloud_directory import DirectoryProjectionInbox, Direc
 from ..entity.persistence.user import AccountSource, AccountStatus, User
 from ..entity.persistence.workspace import (
     MembershipRole,
+    MembershipSource,
     MembershipStatus,
     Workspace,
     WorkspaceExecutionSource,
@@ -358,6 +359,7 @@ class DirectoryProjectionService:
 
         await self._reconcile_entitlement_snapshot_set(snapshot)
         self._publish_runtime_execution_projection(snapshot.workspaces)
+        self._request_model_catalog_sync()
         self._record_batch_cardinality(
             active_workspaces=active_workspace_count,
             workspaces=workspace_count,
@@ -466,6 +468,7 @@ class DirectoryProjectionService:
             returned.values(),
             affected_workspace_uuids=requested,
         )
+        self._request_model_catalog_sync()
         self._record_batch_cardinality(
             active_workspaces=active_workspace_count,
             workspaces=workspace_count,
@@ -475,6 +478,14 @@ class DirectoryProjectionService:
             self._record_success()
         self._consumer_cursor = batch.cursor
 
+    def _request_model_catalog_sync(self) -> None:
+        """Wake model provisioning after a committed directory change."""
+
+        service = getattr(self.ap, 'cloud_model_catalog_service', None)
+        request_sync = getattr(service, 'request_sync', None)
+        if callable(request_sync):
+            request_sync()
+
     def _publish_runtime_execution_projection(
         self,
         workspaces: Iterable[DirectoryWorkspace],
@@ -876,15 +887,15 @@ class DirectoryProjectionService:
                         account_uuid=member.account_uuid,
                         role=role,
                         status=status,
+                        source=MembershipSource.CLOUD_PROJECTION.value,
                         joined_at=joined_at,
                         projection_revision=member.projection_revision,
                     )
                 )
                 continue
-            if membership.projection_revision == 0:
-                # Revision zero is Core-owned collaboration state. Directory
-                # projection seeds memberships, but must not overwrite later
-                # invitation, role, or removal decisions made by Core.
+            if membership.source != MembershipSource.CLOUD_PROJECTION.value:
+                # Core-owned collaboration state is never adopted based on
+                # account provenance, revision, or matching account identity.
                 continue
             if membership.uuid != member.membership_uuid:
                 raise DirectoryProjectionUnavailableError('Directory membership UUID changed for one account')
@@ -896,11 +907,12 @@ class DirectoryProjectionService:
                 raise DirectoryProjectionUnavailableError('Directory membership revision has conflicting contents')
             membership.role = role
             membership.status = status
+            membership.source = MembershipSource.CLOUD_PROJECTION.value
             membership.joined_at = joined_at
             membership.projection_revision = member.projection_revision
 
         for account_uuid, membership in existing.items():
-            if account_uuid not in included_accounts and membership.projection_revision != 0:
+            if account_uuid not in included_accounts and membership.source == MembershipSource.CLOUD_PROJECTION.value:
                 membership.status = MembershipStatus.REMOVED.value
                 membership.projection_revision = max(
                     int(membership.projection_revision),

src/langbot/pkg/cloud/launch.py

@@ -3,11 +3,9 @@ from __future__ import annotations
 import asyncio
 import base64
 import binascii
-import datetime
 import hashlib
 import heapq
 import json
-import math
 import os
 import time
 import typing
@@ -16,10 +14,8 @@ from collections.abc import Callable, Iterable
 from cryptography.exceptions import InvalidSignature
 from cryptography.hazmat.primitives import serialization
 from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey
-import sqlalchemy
-from sqlalchemy.dialects.postgresql import insert as pg_insert
 
-from ..entity.persistence.cloud_directory import SpaceLaunchAssertionConsumption
+from .support_admin import SupportAdminReplayError, SupportAdminSessionError, hash_grant_jti
 
 if typing.TYPE_CHECKING:
     from ..core.app import Application
@@ -27,6 +23,7 @@ if typing.TYPE_CHECKING:
 
 CONTROL_PLANE_TYP = 'langbot-control-plane+jwt'
 LAUNCH_KIND = 'workspace.launch'
+SUPPORT_ADMIN_LAUNCH_KIND = 'workspace.support_admin_launch'
 EXPECTED_ISSUER = 'langbot-space'
 EXPECTED_AUDIENCE = 'langbot-cloud-runtime'
 _CONSUMED_JTI_MAX_ENTRIES = 4096
@@ -125,30 +122,66 @@ class SpaceLaunchService:
         *,
         expected_workspace_uuid: str | None = None,
     ) -> dict[str, str]:
-        claims, clock_skew_seconds = self._verify_assertion(assertion)
+        claims = self._verify_assertion(assertion)
         payload = claims.get('payload')
         if not isinstance(payload, dict):
             raise SpaceLaunchError('Launch assertion payload must be a JSON object')
-        account_uuid = _required_string(payload, 'account_uuid')
+        kind = _required_string(claims, 'kind')
         workspace_uuid = _required_string(payload, 'workspace_uuid')
-        return_path = _required_string(payload, 'return_path')
-        if (
-            not return_path.startswith('/')
-            or return_path.startswith('//')
-            or any(character in return_path for character in ('\\', '\r', '\n', '\t'))
-        ):
-            raise SpaceLaunchError('Launch assertion return path is invalid')
         if expected_workspace_uuid is not None and workspace_uuid != expected_workspace_uuid:
             raise SpaceLaunchError('Launch assertion targets another Workspace')
-        replay_retention_expires_at = _required_int(claims, 'exp', minimum=1) + math.ceil(clock_skew_seconds)
-        await self._consume_jti(_required_string(claims, 'jti'), replay_retention_expires_at)
-        return {
+        if kind == SUPPORT_ADMIN_LAUNCH_KIND:
+            if 'account_uuid' in payload:
+                raise SpaceLaunchError('Admin launch assertion must not identify a customer Account')
+            if payload.get('launch_mode') != 'support_admin' or payload.get('principal_type') != 'support_admin':
+                raise SpaceLaunchError('Admin launch principal must be support_admin')
+            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')
+            grant_jti_hash = hash_grant_jti(_required_string(claims, 'jti'))
+            result = {
+                'workspace_uuid': workspace_uuid,
+                'launch_mode': 'support_admin',
+                'actor_account_uuid': actor_account_uuid,
+                'effective_role': 'owner',
+                'grant_jti_hash': grant_jti_hash,
+            }
+            support_service = getattr(self.ap, 'support_admin_session_service', None)
+            if support_service is None or not callable(getattr(support_service, 'consume_launch_grant', None)):
+                raise SpaceLaunchError('Durable support admin session service is unavailable')
+            try:
+                support_session = await support_service.consume_launch_grant(
+                    grant_jti_hash=grant_jti_hash,
+                    workspace_uuid=workspace_uuid,
+                    actor_account_uuid=actor_account_uuid,
+                )
+            except SupportAdminReplayError as exc:
+                raise SpaceLaunchError('Launch assertion has already been consumed') from exc
+            except SupportAdminSessionError as exc:
+                raise SpaceLaunchError(str(exc)) from exc
+            result['support_admin_token'] = support_session.token
+            self.ap.logger.info(
+                'cloud_support_admin_launch_consumed actor_account_uuid=%s workspace_uuid=%s',
+                result['actor_account_uuid'],
+                workspace_uuid,
+            )
+            return result
+
+        if payload.get('launch_mode') is not None:
+            raise SpaceLaunchError('Launch assertion mode is unsupported')
+        account_uuid = _required_string(payload, 'account_uuid')
+        result = {
             'account_uuid': account_uuid,
             'workspace_uuid': workspace_uuid,
-            'return_path': return_path,
         }
+        await self._consume_jti(_required_string(claims, 'jti'), _required_int(claims, 'exp', minimum=1))
+        return result
 
-    def _verify_assertion(self, token: str) -> tuple[dict[str, typing.Any], float]:
+    def _verify_assertion(self, token: str) -> dict[str, typing.Any]:
         if not getattr(getattr(self.ap, 'deployment', None), 'multi_workspace_enabled', False):
             raise SpaceLaunchError('Space direct launch requires verified Cloud mode')
         public_key, key_id, clock_skew_seconds = self._trust_config()
@@ -184,8 +217,9 @@ class SpaceLaunchService:
             raise SpaceLaunchError('Launch assertion subject targets another instance')
         if _required_string(claims, 'instance_uuid') != instance_uuid:
             raise SpaceLaunchError('Launch assertion instance UUID does not match this Core')
-        if _required_string(claims, 'kind') != LAUNCH_KIND:
-            raise SpaceLaunchError('Launch assertion kind is not workspace.launch')
+        kind = _required_string(claims, 'kind')
+        if kind not in {LAUNCH_KIND, SUPPORT_ADMIN_LAUNCH_KIND}:
+            raise SpaceLaunchError('Launch assertion kind is not supported')
 
         issued_at = _required_int(claims, 'iat')
         not_before = _required_int(claims, 'nbf')
@@ -199,7 +233,7 @@ class SpaceLaunchService:
             raise SpaceLaunchError('Launch assertion is expired')
         if expires_at <= max(issued_at, not_before):
             raise SpaceLaunchError('Launch assertion expiry must follow issue time')
-        return claims, clock_skew_seconds
+        return claims
 
     def _trust_config(self) -> tuple[Ed25519PublicKey, str, float]:
         data = getattr(getattr(self.ap, 'instance_config', None), 'data', {}) or {}
@@ -229,39 +263,19 @@ class SpaceLaunchService:
     async def _consume_jti(self, jti: str, expires_at: int) -> None:
         digest = hashlib.sha256(jti.encode('utf-8')).hexdigest()
         now = int(self._wall_time())
-        persistence_mgr = getattr(self.ap, 'persistence_mgr', None)
-        instance_uuid = str(self.ap.workspace_service.instance_uuid)
-        if persistence_mgr is not None:
-            expires_at_datetime = datetime.datetime.fromtimestamp(expires_at, tz=datetime.timezone.utc)
-            now_datetime = datetime.datetime.fromtimestamp(now, tz=datetime.timezone.utc)
-            async with persistence_mgr.directory_projection_uow(instance_uuid) as uow:
-                await uow.session.execute(
-                    sqlalchemy.delete(SpaceLaunchAssertionConsumption).where(
-                        SpaceLaunchAssertionConsumption.instance_uuid == instance_uuid,
-                        SpaceLaunchAssertionConsumption.expires_at < now_datetime,
-                    )
-                )
-                statement = (
-                    pg_insert(SpaceLaunchAssertionConsumption)
-                    .values(instance_uuid=instance_uuid, jti=digest, expires_at=expires_at_datetime)
-                    .on_conflict_do_nothing(index_elements=['instance_uuid', 'jti'])
-                    .returning(SpaceLaunchAssertionConsumption.jti)
-                )
-                result = await uow.session.execute(statement)
-                if result.scalar_one_or_none() is None:
-                    raise SpaceLaunchError('Launch assertion has already been consumed')
-            return
-
-        # Lightweight unit-test and OSS compatibility fallback. Verified Cloud
-        # runtime always supplies the durable PostgreSQL persistence manager.
         async with self._replay_lock:
             self._prune_consumed_jtis(now)
             if digest in self._consumed_jtis:
                 raise SpaceLaunchError('Launch assertion has already been consumed')
             if len(self._consumed_jtis) >= _CONSUMED_JTI_MAX_ENTRIES:
+                # Evicting a still-valid digest would make a signed launch
+                # assertion replayable. Bound memory by failing closed instead.
                 raise SpaceLaunchError('Launch assertion replay cache capacity reached')
             self._consumed_jtis[digest] = expires_at
-            heapq.heappush(self._consumed_jti_expiry_heap, (expires_at, digest))
+            heapq.heappush(
+                self._consumed_jti_expiry_heap,
+                (expires_at, digest),
+            )
 
     def _prune_consumed_jtis(self, now: int) -> None:
         while self._consumed_jti_expiry_heap:

src/langbot/pkg/cloud/model_catalog.py

@@ -0,0 +1,337 @@
+from __future__ import annotations
+
+import asyncio
+import uuid
+from datetime import datetime
+from typing import Any, Literal, Protocol, runtime_checkable
+
+import sqlalchemy
+from pydantic import BaseModel, ConfigDict, Field, SecretStr, field_validator
+
+from ..entity.persistence import model as persistence_model
+
+
+LANGBOT_MODELS_PROVIDER_REQUESTER = 'space-chat-completions'
+LANGBOT_MODELS_PROVIDER_NAME = 'LangBot Models'
+_MODEL_RESOURCE_NAMESPACE = uuid.UUID('94c703ca-1df5-4e91-bcd3-74ac65cb7921')
+_SUPPORTED_CATEGORIES = {'chat', 'embedding', 'rerank'}
+_MODEL_TABLES = (
+    persistence_model.LLMModel,
+    persistence_model.EmbeddingModel,
+    persistence_model.RerankModel,
+)
+
+
+class CloudModelCatalogItem(BaseModel):
+    model_config = ConfigDict(extra='forbid', frozen=True)
+
+    uuid: str = Field(min_length=1, max_length=255)
+    model_id: str = Field(min_length=1, max_length=255)
+    category: Literal['chat', 'embedding', 'rerank']
+    llm_abilities: tuple[str, ...] = ()
+    is_featured: bool = False
+    featured_order: int = 0
+
+    @field_validator('llm_abilities', mode='before')
+    @classmethod
+    def normalize_missing_abilities(cls, value: Any) -> Any:
+        return () if value is None else value
+
+    @field_validator('llm_abilities')
+    @classmethod
+    def validate_abilities(cls, value: tuple[str, ...]) -> tuple[str, ...]:
+        if any(not item.strip() or len(item) > 64 for item in value):
+            raise ValueError('Model abilities must be non-empty strings of at most 64 characters')
+        if len(set(value)) != len(value):
+            raise ValueError('Model abilities must be unique')
+        return value
+
+
+class CloudWorkspaceModelBilling(BaseModel):
+    model_config = ConfigDict(extra='forbid', frozen=True)
+
+    workspace_uuid: str = Field(min_length=36, max_length=36)
+    owner_account_uuid: str | None = Field(default=None, min_length=36, max_length=36)
+    api_key: SecretStr | None = None
+    credits: int | None = None
+
+    @field_validator('workspace_uuid')
+    @classmethod
+    def validate_uuid(cls, value: str) -> str:
+        return str(uuid.UUID(value))
+
+    @field_validator('owner_account_uuid')
+    @classmethod
+    def validate_optional_uuid(cls, value: str | None) -> str | None:
+        return None if value is None else str(uuid.UUID(value))
+
+
+class CloudModelCatalogSnapshot(BaseModel):
+    model_config = ConfigDict(extra='forbid', frozen=True)
+
+    instance_uuid: str = Field(min_length=1, max_length=255)
+    generated_at: datetime
+    base_url: str = Field(min_length=1, max_length=512)
+    models: tuple[CloudModelCatalogItem, ...]
+    workspaces: tuple[CloudWorkspaceModelBilling, ...]
+
+    @field_validator('base_url')
+    @classmethod
+    def validate_base_url(cls, value: str) -> str:
+        normalized = value.rstrip('/')
+        if not normalized.startswith('https://'):
+            raise ValueError('Cloud model gateway base URL must use HTTPS')
+        return normalized
+
+    @field_validator('models')
+    @classmethod
+    def validate_models(cls, value: tuple[CloudModelCatalogItem, ...]) -> tuple[CloudModelCatalogItem, ...]:
+        if len(value) > 500:
+            raise ValueError('Cloud model catalog exceeds 500 models')
+        identities = {(item.category, item.uuid) for item in value}
+        if len(identities) != len(value):
+            raise ValueError('Cloud model catalog contains duplicate model identities')
+        return value
+
+    @field_validator('workspaces')
+    @classmethod
+    def validate_workspaces(
+        cls, value: tuple[CloudWorkspaceModelBilling, ...]
+    ) -> tuple[CloudWorkspaceModelBilling, ...]:
+        if len(value) > 10_000:
+            raise ValueError('Cloud model catalog exceeds 10000 Workspaces')
+        identities = {item.workspace_uuid for item in value}
+        if len(identities) != len(value):
+            raise ValueError('Cloud model catalog contains duplicate Workspaces')
+        return value
+
+
+@runtime_checkable
+class CloudModelCatalogProvider(Protocol):
+    async def fetch_model_catalog(self, instance_uuid: str) -> CloudModelCatalogSnapshot:
+        """Fetch and verify the complete model catalog and Workspace billing projection."""
+        ...
+
+
+def system_provider_uuid(workspace_uuid: str) -> str:
+    workspace = str(uuid.UUID(workspace_uuid))
+    return str(uuid.uuid5(_MODEL_RESOURCE_NAMESPACE, f'{workspace}:provider:{LANGBOT_MODELS_PROVIDER_REQUESTER}'))
+
+
+def system_model_uuid(workspace_uuid: str, category: str, upstream_uuid: str) -> str:
+    workspace = str(uuid.UUID(workspace_uuid))
+    if category not in _SUPPORTED_CATEGORIES:
+        raise ValueError(f'Unsupported model category: {category}')
+    if not upstream_uuid:
+        raise ValueError('Upstream model UUID is required')
+    return str(uuid.uuid5(_MODEL_RESOURCE_NAMESPACE, f'{workspace}:model:{category}:{upstream_uuid}'))
+
+
+class CloudModelCatalogSyncService:
+    """Reconcile Space-owned model catalog and Owner billing tokens into every Cloud Workspace."""
+
+    def __init__(
+        self,
+        ap: Any,
+        provider: CloudModelCatalogProvider,
+        instance_uuid: str,
+        *,
+        sync_interval_seconds: float = 3600.0,
+    ) -> None:
+        if not isinstance(provider, CloudModelCatalogProvider):
+            raise TypeError('Cloud model catalog sync requires a CloudModelCatalogProvider')
+        if sync_interval_seconds < 10:
+            raise ValueError('Cloud model catalog sync interval must be at least 10 seconds')
+        self.ap = ap
+        self.provider = provider
+        self.instance_uuid = instance_uuid
+        self.sync_interval_seconds = float(sync_interval_seconds)
+        # A tenant UoW commits one Workspace at a time. Keep a durable in-memory
+        # convergence marker so a failed runtime reload is retried even when the
+        # following database reconciliation is a no-op.
+        self._runtime_reload_pending = False
+        self._workspace_credits: dict[str, int | None] = {}
+        self._sync_requested = asyncio.Event()
+
+    def get_workspace_credits(self, workspace_uuid: str) -> int | None:
+        """Return the latest signed owner-credit projection for a Workspace."""
+        return self._workspace_credits.get(str(uuid.UUID(workspace_uuid)))
+
+    async def initialize(self) -> None:
+        await self.sync_once(reload_runtime=False)
+
+    def request_sync(self) -> None:
+        """Wake the catalog loop after a directory Workspace change."""
+
+        self._sync_requested.set()
+
+    async def run(self) -> None:
+        while True:
+            try:
+                await asyncio.wait_for(self._sync_requested.wait(), timeout=self.sync_interval_seconds)
+            except TimeoutError:
+                pass
+            self._sync_requested.clear()
+            try:
+                await self.sync_once(reload_runtime=True)
+            except asyncio.CancelledError:
+                raise
+            except Exception as exc:
+                # Exception messages can contain rendered SQL bound values,
+                # including provider API keys. Log only the exception class.
+                self.ap.logger.warning(f'Cloud model catalog synchronization failed ({type(exc).__name__})')
+
+    async def sync_once(self, *, reload_runtime: bool = True) -> dict[str, int]:
+        summary = {'workspaces': 0, 'created': 0, 'updated': 0, 'deleted': 0}
+        snapshot: CloudModelCatalogSnapshot | None = None
+        sync_error: Exception | None = None
+        reload_error: Exception | None = None
+        try:
+            snapshot = await self.provider.fetch_model_catalog(self.instance_uuid)
+            if snapshot.instance_uuid != self.instance_uuid:
+                raise ValueError('Cloud model catalog targets another LangBot instance')
+
+            bindings = await self.ap.workspace_service.list_active_execution_bindings()
+            billing_by_workspace = {item.workspace_uuid: item for item in snapshot.workspaces}
+            missing = sorted(
+                binding.workspace_uuid for binding in bindings if binding.workspace_uuid not in billing_by_workspace
+            )
+            if missing:
+                raise ValueError(
+                    f'Cloud model catalog is missing billing projections for {len(missing)} active Workspaces'
+                )
+
+            for binding in bindings:
+                counts = await self._sync_workspace(
+                    binding.workspace_uuid,
+                    snapshot,
+                    billing_by_workspace[binding.workspace_uuid],
+                )
+                summary['workspaces'] += 1
+                workspace_changed = any(counts[key] > 0 for key in ('created', 'updated', 'deleted'))
+                if workspace_changed:
+                    # _sync_workspace returns only after its tenant UoW commits.
+                    self._runtime_reload_pending = True
+                for key in ('created', 'updated', 'deleted'):
+                    summary[key] += counts[key]
+                self._workspace_credits[binding.workspace_uuid] = billing_by_workspace[binding.workspace_uuid].credits
+        except Exception as exc:
+            sync_error = exc
+        finally:
+            model_mgr = getattr(self.ap, 'model_mgr', None)
+            if reload_runtime and self._runtime_reload_pending and model_mgr is not None:
+                try:
+                    await model_mgr.load_models_from_db()
+                except Exception as exc:
+                    reload_error = exc
+                else:
+                    self._runtime_reload_pending = False
+
+        if sync_error is not None:
+            if reload_error is not None:
+                raise sync_error from reload_error
+            raise sync_error
+        if reload_error is not None:
+            raise reload_error
+
+        changed = any(summary[key] > 0 for key in ('created', 'updated', 'deleted'))
+        if changed and snapshot is not None:
+            self.ap.logger.info(
+                'Cloud model catalog synchronized '
+                f'({summary["workspaces"]} Workspaces, {len(snapshot.models)} models, '
+                f'created={summary["created"]}, updated={summary["updated"]}, deleted={summary["deleted"]})'
+            )
+        return summary
+
+    async def _sync_workspace(
+        self,
+        workspace_uuid: str,
+        snapshot: CloudModelCatalogSnapshot,
+        billing: CloudWorkspaceModelBilling,
+    ) -> dict[str, int]:
+        counts = {'created': 0, 'updated': 0, 'deleted': 0}
+        provider_uuid = system_provider_uuid(workspace_uuid)
+        desired_keys = [billing.api_key.get_secret_value()] if billing.api_key is not None else []
+
+        async with self.ap.persistence_mgr.tenant_uow(workspace_uuid) as uow:
+            provider = await uow.session.scalar(
+                sqlalchemy.select(persistence_model.ModelProvider).where(
+                    persistence_model.ModelProvider.uuid == provider_uuid
+                )
+            )
+            provider_values = {
+                'workspace_uuid': workspace_uuid,
+                'name': LANGBOT_MODELS_PROVIDER_NAME,
+                'requester': LANGBOT_MODELS_PROVIDER_REQUESTER,
+                'base_url': snapshot.base_url,
+                'api_keys': desired_keys,
+            }
+            if provider is None:
+                provider = persistence_model.ModelProvider(uuid=provider_uuid, **provider_values)
+                uow.session.add(provider)
+                await uow.session.flush()
+                counts['created'] += 1
+            elif self._update_entity(provider, provider_values):
+                counts['updated'] += 1
+
+            existing_by_table: dict[type, dict[str, Any]] = {}
+            for table in _MODEL_TABLES:
+                rows = (
+                    await uow.session.scalars(sqlalchemy.select(table).where(table.provider_uuid == provider_uuid))
+                ).all()
+                existing_by_table[table] = {row.uuid: row for row in rows}
+
+            desired_ids: dict[type, set[str]] = {table: set() for table in _MODEL_TABLES}
+            for item in snapshot.models:
+                table, values = self._model_values(workspace_uuid, provider_uuid, item)
+                model_uuid = system_model_uuid(workspace_uuid, item.category, item.uuid)
+                desired_ids[table].add(model_uuid)
+                existing = existing_by_table[table].get(model_uuid)
+                if existing is None:
+                    uow.session.add(table(uuid=model_uuid, **values))
+                    counts['created'] += 1
+                elif self._update_entity(existing, values):
+                    counts['updated'] += 1
+
+            for table, entities in existing_by_table.items():
+                for model_uuid, entity in entities.items():
+                    if model_uuid not in desired_ids[table]:
+                        await uow.session.delete(entity)
+                        counts['deleted'] += 1
+
+        return counts
+
+    @staticmethod
+    def _update_entity(entity: Any, values: dict[str, Any]) -> bool:
+        changed = False
+        for key, value in values.items():
+            if getattr(entity, key) != value:
+                setattr(entity, key, value)
+                changed = True
+        return changed
+
+    @staticmethod
+    def _model_values(
+        workspace_uuid: str,
+        provider_uuid: str,
+        item: CloudModelCatalogItem,
+    ) -> tuple[type, dict[str, Any]]:
+        ranking = 100 - item.featured_order if item.is_featured else 0
+        common = {
+            'workspace_uuid': workspace_uuid,
+            'name': item.model_id,
+            'provider_uuid': provider_uuid,
+            'extra_args': {},
+            'prefered_ranking': ranking,
+        }
+        if item.category == 'chat':
+            return persistence_model.LLMModel, {
+                **common,
+                'abilities': list(item.llm_abilities),
+                'context_length': None,
+            }
+        if item.category == 'embedding':
+            return persistence_model.EmbeddingModel, common
+        if item.category == 'rerank':
+            return persistence_model.RerankModel, common
+        raise ValueError(f'Unsupported model category: {item.category}')

src/langbot/pkg/cloud/quotas.py

@@ -1,82 +0,0 @@
-from __future__ import annotations
-
-from dataclasses import dataclass
-from typing import Any, Awaitable, Callable
-
-import sqlalchemy
-
-from ..entity.persistence import workspace as persistence_workspace
-from .entitlements import EntitlementResolver
-
-
-Execute = Callable[[Any], Awaitable[Any]]
-
-
-class WorkspaceQuotaExceededError(ValueError):
-    """A stable business error raised when a workspace has no free slots."""
-
-    error_code = 'workspace_quota_exceeded'
-
-    def __init__(self, resource_name: str, limit: int) -> None:
-        self.resource_name = resource_name
-        self.limit = limit
-        super().__init__(f'Maximum number of {resource_name} ({limit}) reached')
-
-
-@dataclass(frozen=True, slots=True)
-class WorkspaceQuota:
-    limit: int
-    requires_transaction_lock: bool
-
-
-async def resolve_workspace_quota(
-    ap: Any,
-    workspace_uuid: str,
-    limit_name: str,
-    *,
-    fallback: int = -1,
-) -> WorkspaceQuota:
-    """Resolve a plan-agnostic Cloud limit while preserving OSS configuration."""
-
-    resolver = getattr(ap, 'entitlement_resolver', None)
-    if isinstance(resolver, EntitlementResolver):
-        snapshot = await resolver.resolve(workspace_uuid)
-        return WorkspaceQuota(
-            limit=snapshot.limit(limit_name),
-            requires_transaction_lock=True,
-        )
-    return WorkspaceQuota(limit=fallback, requires_transaction_lock=False)
-
-
-async def lock_workspace_for_quota(execute: Execute, workspace_uuid: str) -> None:
-    """Serialize quota checks on the durable Workspace row within one transaction."""
-
-    result = await execute(
-        sqlalchemy.select(persistence_workspace.Workspace.uuid)
-        .where(persistence_workspace.Workspace.uuid == workspace_uuid)
-        .with_for_update()
-    )
-    if result.first() is None:
-        raise ValueError('Workspace does not exist')
-
-
-async def require_resource_capacity(
-    execute: Execute,
-    *,
-    workspace_uuid: str,
-    model: type,
-    quota: WorkspaceQuota,
-    resource_name: str,
-    workspace_locked: bool = False,
-) -> None:
-    if quota.limit < 0:
-        return
-    if quota.requires_transaction_lock and not workspace_locked:
-        await lock_workspace_for_quota(execute, workspace_uuid)
-    result = await execute(
-        sqlalchemy.select(sqlalchemy.func.count())
-        .select_from(model)
-        .where(model.workspace_uuid == workspace_uuid)
-    )
-    if int(result.scalar_one()) >= quota.limit:
-        raise WorkspaceQuotaExceededError(resource_name, quota.limit)

src/langbot/pkg/cloud/support_admin.py

@@ -0,0 +1,248 @@
+from __future__ import annotations
+
+import dataclasses
+import datetime
+import hashlib
+import re
+import time
+import typing
+
+import jwt
+from sqlalchemy.exc import IntegrityError
+
+from ..entity.persistence.support_admin import SupportAdminTemporarySession
+from ..workspace.errors import WorkspaceError
+
+if typing.TYPE_CHECKING:
+    from ..core.app import Application
+
+
+SUPPORT_ADMIN_TOKEN_TYP = 'langbot-support-admin+jwt'
+SUPPORT_ADMIN_TOKEN_KIND = 'support_admin.session'
+SUPPORT_ADMIN_EFFECTIVE_ROLE = 'owner'
+SUPPORT_ADMIN_MAX_TOKEN_SECONDS = 300
+_SHA256_HEX = re.compile(r'^[0-9a-f]{64}$')
+
+
+class SupportAdminSessionError(ValueError):
+    """Raised when a support-admin session or token is not admissible."""
+
+
+class SupportAdminReplayError(SupportAdminSessionError):
+    """Raised when a launch grant JTI has already been consumed."""
+
+
+@dataclasses.dataclass(frozen=True, slots=True)
+class IssuedSupportAdminSession:
+    token: str
+    grant_jti_hash: str
+    workspace_uuid: str
+    actor_account_uuid: str
+    issued_at: datetime.datetime
+    expires_at: datetime.datetime
+
+
+@dataclasses.dataclass(frozen=True, slots=True)
+class SupportAdminSessionIdentity:
+    grant_jti_hash: str
+    workspace_uuid: str
+    actor_account_uuid: str
+    instance_uuid: str
+    placement_generation: int
+
+
+def hash_grant_jti(jti: str) -> str:
+    return hashlib.sha256(jti.encode('utf-8')).hexdigest()
+
+
+class SupportAdminSessionService:
+    """Issue and validate temporary Workspace-scoped support-admin sessions."""
+
+    def __init__(
+        self,
+        ap: Application,
+        *,
+        wall_time: typing.Callable[[], float] = time.time,
+    ) -> None:
+        self.ap = ap
+        self._wall_time = wall_time
+
+    async def consume_launch_grant(
+        self,
+        *,
+        grant_jti_hash: str,
+        workspace_uuid: str,
+        actor_account_uuid: str,
+    ) -> IssuedSupportAdminSession:
+        self._validate_grant_hash(grant_jti_hash)
+        if not workspace_uuid or not actor_account_uuid:
+            raise SupportAdminSessionError('Support admin session requires an actor and Workspace')
+
+        issued_at = self._utcnow()
+        expires_at = issued_at + datetime.timedelta(seconds=SUPPORT_ADMIN_MAX_TOKEN_SECONDS)
+        tenant_uow = getattr(self.ap.persistence_mgr, 'tenant_uow', None)
+        if not callable(tenant_uow):
+            raise SupportAdminSessionError('Support admin sessions require tenant persistence')
+
+        try:
+            async with tenant_uow(workspace_uuid) as uow:
+                await self.ap.workspace_service.get_execution_binding(workspace_uuid, session=uow.session)
+                uow.session.add(
+                    SupportAdminTemporarySession(
+                        grant_jti_hash=grant_jti_hash,
+                        workspace_uuid=workspace_uuid,
+                        actor_account_uuid=actor_account_uuid,
+                        issued_at=issued_at,
+                        expires_at=expires_at,
+                    )
+                )
+                await uow.session.flush()
+        except IntegrityError as exc:
+            raise SupportAdminReplayError('Launch assertion has already been consumed') from exc
+        except WorkspaceError as exc:
+            raise SupportAdminSessionError('Workspace is unavailable for support access') from exc
+
+        return IssuedSupportAdminSession(
+            token=self._encode_token(
+                grant_jti_hash=grant_jti_hash,
+                workspace_uuid=workspace_uuid,
+                actor_account_uuid=actor_account_uuid,
+                issued_at=issued_at,
+                expires_at=expires_at,
+            ),
+            grant_jti_hash=grant_jti_hash,
+            workspace_uuid=workspace_uuid,
+            actor_account_uuid=actor_account_uuid,
+            issued_at=issued_at,
+            expires_at=expires_at,
+        )
+
+    def is_support_admin_token(self, token: str) -> bool:
+        """Return True only for compact JWTs marked as support-admin tokens."""
+
+        if not isinstance(token, str) or token.count('.') != 2:
+            return False
+        try:
+            header = jwt.get_unverified_header(token)
+        except jwt.PyJWTError:
+            return False
+        if header.get('typ') == SUPPORT_ADMIN_TOKEN_TYP:
+            return True
+        try:
+            payload = jwt.decode(token, options={'verify_signature': False})
+        except jwt.PyJWTError:
+            return False
+        return payload.get('kind') == SUPPORT_ADMIN_TOKEN_KIND
+
+    async def authenticate_token(
+        self,
+        token: str,
+        *,
+        requested_workspace_uuid: str | None,
+    ) -> SupportAdminSessionIdentity:
+        if not self.is_support_admin_token(token):
+            raise SupportAdminSessionError('Not a support admin token')
+        workspace_uuid = (requested_workspace_uuid or '').strip()
+        if not workspace_uuid:
+            raise SupportAdminSessionError('Support admin token requires an explicit Workspace selector')
+
+        jwt_secret = self.ap.instance_config.data['system']['jwt']['secret']
+        try:
+            payload = jwt.decode(
+                token,
+                jwt_secret,
+                algorithms=['HS256'],
+                issuer='langbot-core',
+                audience=self._audience(workspace_uuid),
+                options={'require': ['exp', 'iat', 'nbf', 'iss', 'aud']},
+            )
+        except jwt.PyJWTError as exc:
+            raise SupportAdminSessionError('Invalid support admin token') from exc
+        self._validate_payload(payload, workspace_uuid)
+        grant_jti_hash = payload['grant_jti_hash']
+        actor_account_uuid = payload['actor_account_uuid']
+
+        tenant_uow = getattr(self.ap.persistence_mgr, 'tenant_uow', None)
+        if not callable(tenant_uow):
+            raise SupportAdminSessionError('Support admin sessions require tenant persistence')
+
+        now = self._utcnow()
+        async with tenant_uow(workspace_uuid) as uow:
+            session = await uow.session.get(SupportAdminTemporarySession, grant_jti_hash)
+            if (
+                session is None
+                or session.workspace_uuid != workspace_uuid
+                or session.actor_account_uuid != actor_account_uuid
+                or session.revoked_at is not None
+                or session.expires_at <= now
+            ):
+                raise SupportAdminSessionError('Support admin session is inactive')
+            binding = await self.ap.workspace_service.get_execution_binding(workspace_uuid, session=uow.session)
+            session.last_used_at = now
+            await uow.session.flush()
+
+        return SupportAdminSessionIdentity(
+            grant_jti_hash=grant_jti_hash,
+            workspace_uuid=workspace_uuid,
+            actor_account_uuid=actor_account_uuid,
+            instance_uuid=binding.instance_uuid,
+            placement_generation=binding.placement_generation,
+        )
+
+    async def revoke_session(self, grant_jti_hash: str, workspace_uuid: str) -> None:
+        self._validate_grant_hash(grant_jti_hash)
+        now = self._utcnow()
+        async with self.ap.persistence_mgr.tenant_uow(workspace_uuid) as uow:
+            row = await uow.session.get(SupportAdminTemporarySession, grant_jti_hash)
+            if row is not None and row.revoked_at is None:
+                row.revoked_at = now
+
+    def _encode_token(
+        self,
+        *,
+        grant_jti_hash: str,
+        workspace_uuid: str,
+        actor_account_uuid: str,
+        issued_at: datetime.datetime,
+        expires_at: datetime.datetime,
+    ) -> str:
+        jwt_secret = self.ap.instance_config.data['system']['jwt']['secret']
+        payload: dict[str, typing.Any] = {
+            'kind': SUPPORT_ADMIN_TOKEN_KIND,
+            'iss': 'langbot-core',
+            'aud': self._audience(workspace_uuid),
+            'sub': f'support-admin:{actor_account_uuid}',
+            'iat': issued_at,
+            'nbf': issued_at,
+            'exp': expires_at,
+            'actor_account_uuid': actor_account_uuid,
+            'workspace_uuid': workspace_uuid,
+            'effective_role': SUPPORT_ADMIN_EFFECTIVE_ROLE,
+            'grant_jti_hash': grant_jti_hash,
+        }
+        return jwt.encode(payload, jwt_secret, algorithm='HS256', headers={'typ': SUPPORT_ADMIN_TOKEN_TYP})
+
+    def _validate_payload(self, payload: dict[str, typing.Any], workspace_uuid: str) -> None:
+        if payload.get('kind') != SUPPORT_ADMIN_TOKEN_KIND:
+            raise SupportAdminSessionError('Invalid support admin token kind')
+        if payload.get('workspace_uuid') != workspace_uuid:
+            raise SupportAdminSessionError('Support admin session is scoped to another Workspace')
+        if payload.get('effective_role') != SUPPORT_ADMIN_EFFECTIVE_ROLE:
+            raise SupportAdminSessionError('Invalid support admin token role')
+        actor_account_uuid = payload.get('actor_account_uuid')
+        if not isinstance(actor_account_uuid, str) or not actor_account_uuid.strip():
+            raise SupportAdminSessionError('Invalid support admin actor')
+        grant_jti_hash = payload.get('grant_jti_hash')
+        if not isinstance(grant_jti_hash, str) or not _SHA256_HEX.match(grant_jti_hash):
+            raise SupportAdminSessionError('Invalid support admin grant')
+
+    def _audience(self, workspace_uuid: str) -> str:
+        return f'langbot-support-admin:{self.ap.workspace_service.instance_uuid}:{workspace_uuid}'
+
+    @staticmethod
+    def _validate_grant_hash(grant_jti_hash: str) -> None:
+        if not _SHA256_HEX.match(grant_jti_hash):
+            raise SupportAdminSessionError('Invalid support admin grant')
+
+    def _utcnow(self) -> datetime.datetime:
+        return datetime.datetime.fromtimestamp(self._wall_time(), datetime.UTC).replace(tzinfo=None)

src/langbot/pkg/core/app.py

@@ -51,8 +51,10 @@ from ..workspace import collaboration as workspace_collaboration_module
 from ..workspace import invitation_delivery as invitation_delivery_module
 from ..cloud import bootstrap as cloud_bootstrap_module
 from ..cloud import launch as cloud_launch_module
+from ..cloud import support_admin as cloud_support_admin_module
 from ..cloud import directory_projection as cloud_directory_projection_module
 from ..cloud import entitlements as cloud_entitlements_module
+from ..cloud import model_catalog as cloud_model_catalog_module
 from ..api.http.context import ExecutionContext, PrincipalContext, PrincipalType
 
 
@@ -136,16 +138,17 @@ class Application:
 
     space_launch_service: cloud_launch_module.SpaceLaunchService = None
 
+    support_admin_session_service: cloud_support_admin_module.SupportAdminSessionService = None
+
     deployment: cloud_bootstrap_module.OpenSourceDeployment | cloud_bootstrap_module.VerifiedCloudDeployment = None
 
     deployment_admission: cloud_bootstrap_module.DeploymentAdmissionGuard = None
-
+    directory_projection_service: cloud_directory_projection_module.DirectoryProjectionService | None = None
+    cloud_model_catalog_service: cloud_model_catalog_module.CloudModelCatalogSyncService | None = None
     manifest_refresh_service: cloud_bootstrap_module.CloudManifestRefreshService | None = None
 
     entitlement_resolver: cloud_entitlements_module.EntitlementResolver | None = None
 
-    directory_projection_service: cloud_directory_projection_module.DirectoryProjectionService | None = None
-
     vector_db_mgr: vectordb_mgr.VectorDBManager = None
 
     http_ctrl: http_controller.HTTPController = None
@@ -303,6 +306,12 @@ class Application:
                     name='cloud-directory-projection',
                     scopes=[core_entities.LifecycleControlScope.APPLICATION],
                 )
+            if self.cloud_model_catalog_service is not None:
+                self.task_mgr.create_task(
+                    self.cloud_model_catalog_service.run(),
+                    name='cloud-model-catalog-sync',
+                    scopes=[core_entities.LifecycleControlScope.APPLICATION],
+                )
             if self.manifest_refresh_service is not None:
                 self.task_mgr.create_task(
                     self.manifest_refresh_service.run(),

src/langbot/pkg/core/stages/build_app.py

@@ -42,9 +42,11 @@ from ...workspace import collaboration as workspace_collaboration_module
 from ...workspace import invitation_delivery as invitation_delivery_module
 from ...cloud import bootstrap as cloud_bootstrap
 from ...cloud import launch as cloud_launch_module
+from ...cloud import support_admin as cloud_support_admin_module
 from ...cloud.directory import directory_projection_limits_from_config
 from ...cloud.directory_projection import DirectoryProjectionService
 from ...cloud.entitlements import EntitlementResolver
+from ...cloud.model_catalog import CloudModelCatalogSyncService
 from ...api.http.context import ExecutionContext, PrincipalContext, PrincipalType
 from ...api.http.authz import WorkspaceRequiredError
 
@@ -175,11 +177,22 @@ class BuildAppStage(stage.BootingStage):
             # of repeating tenant validation for every manager.
             await workspace_service_inst.prime_startup_execution_bindings()
 
+            if not isinstance(deployment, cloud_bootstrap.VerifiedCloudDeployment):
+                raise RuntimeError('Multi-Workspace runtime requires a verified Cloud deployment')
+            cloud_model_catalog_service = CloudModelCatalogSyncService(
+                ap,
+                deployment.model_catalog_provider,
+                constants.instance_id,
+            )
+            await cloud_model_catalog_service.initialize()
+            ap.cloud_model_catalog_service = cloud_model_catalog_service
+
         ap.workspace_collaboration_service = workspace_collaboration_module.WorkspaceCollaborationService(
             ap,
             workspace_service_inst,
         )
         ap.invitation_delivery_service = invitation_delivery_module.InvitationDeliveryService(ap)
+        ap.support_admin_session_service = cloud_support_admin_module.SupportAdminSessionService(ap)
         ap.space_launch_service = cloud_launch_module.SpaceLaunchService(ap)
 
         user_service_inst = user_service.UserService(ap)

src/langbot/pkg/core/stages/load_config.py

@@ -41,6 +41,7 @@ _RUNTIME_POLICY_DEFAULTS = {
         }
     },
     'plugin': {
+        'connect_timeout_seconds': 180.0,
         'worker': {
             'max_cpus': 1.0,
             'max_memory_mb': 512,
@@ -56,7 +57,7 @@ _RUNTIME_POLICY_DEFAULTS = {
             'restart_failure_window_seconds': 30.0,
             'restart_circuit_open_seconds': 60.0,
             'require_hard_limits': False,
-        }
+        },
     },
     'mcp': {'stdio': {'enabled': True}},
     'monitoring': {

src/langbot/pkg/entity/persistence/cloud_directory.py

@@ -67,26 +67,3 @@ class DirectoryProjectionInbox(Base):
             name='ck_directory_projection_inbox_fingerprint',
         ),
     )
-
-
-class SpaceLaunchAssertionConsumption(Base):
-    """Durable, instance-scoped replay ledger for signed Space launch assertions."""
-
-    __tablename__ = 'space_launch_assertion_consumptions'
-
-    instance_uuid = sqlalchemy.Column(sqlalchemy.String(255), primary_key=True)
-    jti = sqlalchemy.Column(sqlalchemy.String(255), primary_key=True)
-    expires_at = sqlalchemy.Column(sqlalchemy.DateTime(timezone=True), nullable=False)
-    consumed_at = sqlalchemy.Column(
-        sqlalchemy.DateTime(timezone=True),
-        nullable=False,
-        server_default=sqlalchemy.func.now(),
-    )
-
-    __table_args__ = (
-        sqlalchemy.Index(
-            'ix_space_launch_assertion_consumptions_expiry',
-            'instance_uuid',
-            'expires_at',
-        ),
-    )

src/langbot/pkg/entity/persistence/support_admin.py

@@ -0,0 +1,35 @@
+from __future__ import annotations
+
+import sqlalchemy
+
+from .base import Base
+
+
+class SupportAdminTemporarySession(Base):
+    """Temporary support-admin Workspace access session."""
+
+    __tablename__ = 'support_admin_temporary_sessions'
+
+    grant_jti_hash = sqlalchemy.Column(sqlalchemy.String(64), primary_key=True)
+    workspace_uuid = sqlalchemy.Column(
+        sqlalchemy.String(36),
+        sqlalchemy.ForeignKey('workspaces.uuid', ondelete='CASCADE'),
+        nullable=False,
+    )
+    actor_account_uuid = sqlalchemy.Column(sqlalchemy.String(36), nullable=False)
+    issued_at = sqlalchemy.Column(sqlalchemy.DateTime, nullable=False)
+    expires_at = sqlalchemy.Column(sqlalchemy.DateTime, nullable=False)
+    revoked_at = sqlalchemy.Column(sqlalchemy.DateTime, nullable=True)
+    last_used_at = sqlalchemy.Column(sqlalchemy.DateTime, nullable=True)
+
+    __table_args__ = (
+        sqlalchemy.Index(
+            'ix_support_admin_sessions_workspace_expiry',
+            'workspace_uuid',
+            'expires_at',
+        ),
+        sqlalchemy.CheckConstraint(
+            'length(grant_jti_hash) = 64',
+            name='ck_support_admin_sessions_grant_jti_hash',
+        ),
+    )

src/langbot/pkg/entity/persistence/workspace.py

@@ -40,6 +40,11 @@ class MembershipStatus(enum.StrEnum):
     REMOVED = 'removed'
 
 
+class MembershipSource(enum.StrEnum):
+    LOCAL = 'local'
+    CLOUD_PROJECTION = 'cloud_projection'
+
+
 class InvitationStatus(enum.StrEnum):
     PENDING = 'pending'
     ACCEPTED = 'accepted'
@@ -151,6 +156,11 @@ class WorkspaceMembership(Base):
         nullable=True,
     )
     joined_at = sqlalchemy.Column(sqlalchemy.DateTime, nullable=True)
+    source = sqlalchemy.Column(
+        sqlalchemy.String(32),
+        nullable=False,
+        server_default=MembershipSource.LOCAL.value,
+    )
     projection_revision = sqlalchemy.Column(sqlalchemy.BigInteger, nullable=False, server_default='0')
     created_at = sqlalchemy.Column(sqlalchemy.DateTime, nullable=False, server_default=sqlalchemy.func.now())
     updated_at = sqlalchemy.Column(
@@ -163,6 +173,13 @@ class WorkspaceMembership(Base):
     __table_args__ = (
         sqlalchemy.UniqueConstraint('workspace_uuid', 'account_uuid', name='uq_workspace_membership_account'),
         sqlalchemy.Index('ix_workspace_memberships_account_status', 'account_uuid', 'status'),
+        sqlalchemy.Index(
+            'uq_workspace_memberships_one_active_owner',
+            'workspace_uuid',
+            unique=True,
+            sqlite_where=sqlalchemy.text("role = 'owner' AND status = 'active'"),
+            postgresql_where=sqlalchemy.text("role = 'owner' AND status = 'active'"),
+        ),
         sqlalchemy.CheckConstraint(
             "role IN ('owner', 'admin', 'developer', 'operator', 'viewer')",
             name='ck_workspace_memberships_role',
@@ -171,6 +188,10 @@ class WorkspaceMembership(Base):
             "status IN ('active', 'disabled', 'removed')",
             name='ck_workspace_memberships_status',
         ),
+        sqlalchemy.CheckConstraint(
+            "source IN ('local', 'cloud_projection')",
+            name='ck_workspace_memberships_source',
+        ),
     )
 
 

src/langbot/pkg/persistence/alembic/versions/0009_workspace_tenancy_kernel.py

@@ -18,6 +18,17 @@ down_revision = '0008_mcp_resource_prefs'
 branch_labels = None
 depends_on = None
 
+_WORKSPACE_IDENTITY_NAMESPACE = uuid.UUID('8ea04f29-8528-4cc3-bb28-30a838c89d76')
+
+
+def _workspace_uuid_from_instance_id(instance_id: str) -> str:
+    value = instance_id.strip()
+    candidate = value[len('instance_') :] if value.startswith('instance_') else value
+    try:
+        return str(uuid.UUID(candidate))
+    except ValueError:
+        return str(uuid.uuid5(_WORKSPACE_IDENTITY_NAMESPACE, value))
+
 
 def _table_names(conn: sa.Connection) -> set[str]:
     return set(sa.inspect(conn).get_table_names())
@@ -403,7 +414,7 @@ def _bootstrap_default_workspace(conn: sa.Connection) -> None:
                 .values(created_by_account_uuid=owner_account_uuid)
             )
     else:
-        workspace_uuid = str(uuid.uuid4())
+        workspace_uuid = _workspace_uuid_from_instance_id(instance_uuid)
         conn.execute(
             workspaces.insert().values(
                 uuid=workspace_uuid,

src/langbot/pkg/persistence/alembic/versions/0016_support_admin_sessions.py

@@ -0,0 +1,88 @@
+"""add temporary support-admin sessions
+
+Revision ID: 0016_support_admin_sessions
+Revises: 0015_cloud_core_collab
+Create Date: 2026-07-31
+"""
+
+from __future__ import annotations
+
+import sqlalchemy as sa
+from alembic import op
+
+
+revision = '0016_support_admin_sessions'
+down_revision = '0015_cloud_core_collab'
+branch_labels = None
+depends_on = None
+
+
+_TABLE_NAME = 'support_admin_temporary_sessions'
+_POLICY_NAME = 'langbot_workspace_isolation'
+_TENANT_SETTING = 'langbot.workspace_uuid'
+
+
+def _setting(name: str) -> str:
+    return f"NULLIF(current_setting('{name}', true), '')"
+
+
+def _quote(conn: sa.Connection, identifier: str) -> str:
+    return conn.dialect.identifier_preparer.quote(identifier)
+
+
+def upgrade() -> None:
+    conn = op.get_bind()
+    existing_tables = set(sa.inspect(conn).get_table_names())
+    if _TABLE_NAME not in existing_tables:
+        op.create_table(
+            _TABLE_NAME,
+            sa.Column('grant_jti_hash', sa.String(64), nullable=False),
+            sa.Column(
+                'workspace_uuid',
+                sa.String(36),
+                sa.ForeignKey('workspaces.uuid', ondelete='CASCADE'),
+                nullable=False,
+            ),
+            sa.Column('actor_account_uuid', sa.String(36), nullable=False),
+            sa.Column('issued_at', sa.DateTime(), nullable=False),
+            sa.Column('expires_at', sa.DateTime(), nullable=False),
+            sa.Column('revoked_at', sa.DateTime(), nullable=True),
+            sa.Column('last_used_at', sa.DateTime(), nullable=True),
+            sa.CheckConstraint(
+                'length(grant_jti_hash) = 64',
+                name='ck_support_admin_sessions_grant_jti_hash',
+            ),
+            sa.PrimaryKeyConstraint('grant_jti_hash'),
+        )
+        op.create_index(
+            'ix_support_admin_sessions_workspace_expiry',
+            _TABLE_NAME,
+            ['workspace_uuid', 'expires_at'],
+            unique=False,
+        )
+
+    if conn.dialect.name != 'postgresql':
+        return
+
+    table = _quote(conn, _TABLE_NAME)
+    policy = _quote(conn, _POLICY_NAME)
+    expression = f'workspace_uuid::text = {_setting(_TENANT_SETTING)}'
+    op.execute(sa.text(f'ALTER TABLE {table} ENABLE ROW LEVEL SECURITY'))
+    op.execute(sa.text(f'ALTER TABLE {table} FORCE ROW LEVEL SECURITY'))
+    op.execute(sa.text(f'DROP POLICY IF EXISTS {policy} ON {table}'))
+    op.execute(
+        sa.text(
+            f'CREATE POLICY {policy} ON {table} AS PERMISSIVE FOR ALL TO PUBLIC '
+            f'USING ({expression}) WITH CHECK ({expression})'
+        )
+    )
+
+
+def downgrade() -> None:
+    conn = op.get_bind()
+    if conn.dialect.name == 'postgresql':
+        table = _quote(conn, _TABLE_NAME)
+        policy = _quote(conn, _POLICY_NAME)
+        op.execute(sa.text(f'DROP POLICY IF EXISTS {policy} ON {table}'))
+    op.drop_index('ix_support_admin_sessions_workspace_expiry', table_name=_TABLE_NAME)
+    op.drop_table(_TABLE_NAME)

src/langbot/pkg/persistence/alembic/versions/0017_oss_workspace_identity.py

@@ -0,0 +1,167 @@
+"""align the OSS Workspace UUID with the persisted instance identity
+
+Revision ID: 0017_oss_workspace_identity
+Revises: 0016_support_admin_sessions
+Create Date: 2026-07-31
+"""
+
+from __future__ import annotations
+
+import uuid
+
+import sqlalchemy as sa
+from alembic import op
+
+
+revision = '0017_oss_workspace_identity'
+down_revision = '0016_support_admin_sessions'
+branch_labels = None
+depends_on = None
+
+_WORKSPACE_IDENTITY_NAMESPACE = uuid.UUID('8ea04f29-8528-4cc3-bb28-30a838c89d76')
+_OSS_WORKSPACE_METADATA_KEY = 'oss_workspace_uuid'
+
+
+def _workspace_uuid_from_instance_id(instance_id: str) -> str:
+    value = instance_id.strip()
+    candidate = value[len('instance_') :] if value.startswith('instance_') else value
+    try:
+        return str(uuid.UUID(candidate))
+    except ValueError:
+        return str(uuid.uuid5(_WORKSPACE_IDENTITY_NAMESPACE, value))
+
+
+def _quote(conn: sa.Connection, identifier: str) -> str:
+    return conn.dialect.identifier_preparer.quote(identifier)
+
+
+def _defer_foreign_keys(conn: sa.Connection, inspector: sa.Inspector, table_names: list[str]) -> None:
+    """Allow the transaction to re-key a connected tenant graph atomically."""
+
+    if conn.dialect.name == 'sqlite':
+        conn.execute(sa.text('PRAGMA defer_foreign_keys = ON'))
+        return
+    if conn.dialect.name != 'postgresql':
+        raise RuntimeError(f'Unsupported Workspace identity migration dialect: {conn.dialect.name}')
+
+    for table_name in table_names:
+        for foreign_key in inspector.get_foreign_keys(table_name):
+            constraint_name = foreign_key.get('name')
+            if not constraint_name:
+                continue
+            conn.execute(
+                sa.text(
+                    f'ALTER TABLE {_quote(conn, table_name)} '
+                    f'ALTER CONSTRAINT {_quote(conn, constraint_name)} DEFERRABLE INITIALLY DEFERRED'
+                )
+            )
+
+
+def _suspend_postgres_rls(
+    conn: sa.Connection,
+    table_names: list[str],
+) -> dict[str, tuple[bool, bool]]:
+    if conn.dialect.name != 'postgresql':
+        return {}
+
+    states: dict[str, tuple[bool, bool]] = {}
+    for table_name in table_names:
+        row = conn.execute(
+            sa.text('SELECT relrowsecurity, relforcerowsecurity FROM pg_class WHERE oid = to_regclass(:table_name)'),
+            {'table_name': table_name},
+        ).one()
+        enabled, forced = bool(row.relrowsecurity), bool(row.relforcerowsecurity)
+        states[table_name] = (enabled, forced)
+        table = _quote(conn, table_name)
+        if forced:
+            conn.execute(sa.text(f'ALTER TABLE {table} NO FORCE ROW LEVEL SECURITY'))
+        if enabled:
+            conn.execute(sa.text(f'ALTER TABLE {table} DISABLE ROW LEVEL SECURITY'))
+    return states
+
+
+def _restore_postgres_rls(conn: sa.Connection, states: dict[str, tuple[bool, bool]]) -> None:
+    for table_name, (enabled, forced) in states.items():
+        table = _quote(conn, table_name)
+        if enabled:
+            conn.execute(sa.text(f'ALTER TABLE {table} ENABLE ROW LEVEL SECURITY'))
+        if forced:
+            conn.execute(sa.text(f'ALTER TABLE {table} FORCE ROW LEVEL SECURITY'))
+
+
+def upgrade() -> None:
+    conn = op.get_bind()
+    inspector = sa.inspect(conn)
+    table_names = inspector.get_table_names()
+    if 'workspaces' not in table_names:
+        return
+
+    metadata = sa.MetaData()
+    workspaces = sa.Table('workspaces', metadata, autoload_with=conn)
+    local_rows = conn.execute(sa.select(workspaces).where(workspaces.c.source == 'local')).mappings().all()
+    if not local_rows:
+        return
+    if len(local_rows) != 1:
+        raise RuntimeError('Cannot align OSS Workspace identity: expected exactly one local Workspace')
+
+    old_row = dict(local_rows[0])
+    old_uuid = old_row['uuid']
+    canonical_uuid = _workspace_uuid_from_instance_id(old_row['instance_uuid'])
+    if old_uuid == canonical_uuid:
+        return
+    if conn.execute(sa.select(workspaces.c.uuid).where(workspaces.c.uuid == canonical_uuid)).scalar_one_or_none():
+        raise RuntimeError(f'Cannot align OSS Workspace identity: target {canonical_uuid!r} already exists')
+
+    tenant_tables = [
+        table_name
+        for table_name in table_names
+        if table_name == 'workspaces'
+        or 'workspace_uuid' in {column['name'] for column in inspector.get_columns(table_name)}
+    ]
+    rls_states = _suspend_postgres_rls(conn, tenant_tables)
+    try:
+        _defer_foreign_keys(conn, inspector, table_names)
+
+        # Release local source/slug uniqueness while the canonical parent exists
+        # alongside the old parent for the duration of this transaction.
+        temporary_slug = f'__workspace_rekey__{old_uuid}'
+        conn.execute(
+            workspaces.update()
+            .where(workspaces.c.uuid == old_uuid)
+            .values(source='cloud_projection', slug=temporary_slug)
+        )
+        new_row = dict(old_row)
+        new_row['uuid'] = canonical_uuid
+        conn.execute(workspaces.insert().values(**new_row))
+
+        for table_name in tenant_tables:
+            if table_name == 'workspaces':
+                continue
+            table = sa.Table(table_name, metadata, autoload_with=conn, extend_existing=True)
+            conn.execute(table.update().where(table.c.workspace_uuid == old_uuid).values(workspace_uuid=canonical_uuid))
+
+        if 'metadata' in table_names:
+            conn.execute(
+                sa.text('UPDATE metadata SET value = :canonical_uuid WHERE key = :key AND value = :old_uuid'),
+                {
+                    'canonical_uuid': canonical_uuid,
+                    'key': _OSS_WORKSPACE_METADATA_KEY,
+                    'old_uuid': old_uuid,
+                },
+            )
+        conn.execute(workspaces.delete().where(workspaces.c.uuid == old_uuid))
+        if conn.dialect.name == 'postgresql':
+            # Fire deferred FK triggers before ALTER TABLE restores RLS; PostgreSQL
+            # rejects ALTER TABLE while a relation has pending trigger events.
+            conn.execute(sa.text('SET CONSTRAINTS ALL IMMEDIATE'))
+    except Exception:
+        # Alembic owns the transaction. Rollback restores the transactional RLS DDL.
+        raise
+    else:
+        _restore_postgres_rls(conn, rls_states)
+
+
+def downgrade() -> None:
+    # The previous random UUID is intentionally not recoverable. Keeping the
+    # canonical identity preserves every FK and is safe for older application code.
+    pass

src/langbot/pkg/persistence/alembic/versions/0018_merge_launch_replay.py

@@ -0,0 +1,21 @@
+"""merge the published Space launch replay and main migration branches
+
+Revision ID: 0018_merge_launch_replay
+Revises: 0016_space_launch_replay, 0017_oss_workspace_identity
+Create Date: 2026-08-01
+"""
+
+from __future__ import annotations
+
+revision = '0018_merge_launch_replay'
+down_revision = ('0016_space_launch_replay', '0017_oss_workspace_identity')
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    pass
+
+
+def downgrade() -> None:
+    pass

src/langbot/pkg/persistence/alembic/versions/0019_single_workspace_owner.py

@@ -0,0 +1,80 @@
+"""enforce one active owner per Workspace
+
+Revision ID: 0019_single_workspace_owner
+Revises: 0018_merge_launch_replay
+Create Date: 2026-08-02
+"""
+
+from __future__ import annotations
+
+import sqlalchemy as sa
+from alembic import op
+
+revision = '0019_single_workspace_owner'
+down_revision = '0018_merge_launch_replay'
+branch_labels = None
+depends_on = None
+
+_INDEX_NAME = 'uq_workspace_memberships_one_active_owner'
+
+
+def upgrade() -> None:
+    conn = op.get_bind()
+    inspector = sa.inspect(conn)
+    if 'workspace_memberships' not in inspector.get_table_names():
+        return
+
+    # Ownership transfer used to promote a second member without demoting the
+    # original owner. Preserve the Workspace creator where possible and demote
+    # every historical extra owner before installing the database invariant.
+    op.execute(
+        sa.text(
+            """
+            WITH ranked_owners AS (
+                SELECT membership.uuid,
+                       ROW_NUMBER() OVER (
+                           PARTITION BY membership.workspace_uuid
+                           ORDER BY
+                               CASE
+                                   WHEN membership.account_uuid = workspace.created_by_account_uuid THEN 0
+                                   ELSE 1
+                               END,
+                               COALESCE(membership.joined_at, membership.created_at),
+                               membership.uuid
+                       ) AS owner_rank
+                FROM workspace_memberships AS membership
+                JOIN workspaces AS workspace
+                  ON workspace.uuid = membership.workspace_uuid
+                WHERE membership.role = 'owner'
+                  AND membership.status = 'active'
+            )
+            UPDATE workspace_memberships
+               SET role = 'admin'
+             WHERE uuid IN (
+                 SELECT uuid
+                   FROM ranked_owners
+                  WHERE owner_rank > 1
+             )
+            """
+        )
+    )
+    # Fresh installations may already have this index because SQLAlchemy
+    # metadata is created before Alembic advances the revision marker.
+    op.execute(
+        sa.text(
+            'CREATE UNIQUE INDEX IF NOT EXISTS '
+            'uq_workspace_memberships_one_active_owner '
+            'ON workspace_memberships (workspace_uuid) '
+            "WHERE role = 'owner' AND status = 'active'"
+        )
+    )
+
+
+def downgrade() -> None:
+    conn = op.get_bind()
+    inspector = sa.inspect(conn)
+    if 'workspace_memberships' not in inspector.get_table_names():
+        return
+    index_names = {index['name'] for index in inspector.get_indexes('workspace_memberships')}
+    if _INDEX_NAME in index_names:
+        op.drop_index(_INDEX_NAME, table_name='workspace_memberships')

src/langbot/pkg/persistence/alembic/versions/001a_pgvector_dimension_3072.py

@@ -0,0 +1,43 @@
+"""enable 3072-dimensional pgvector embeddings
+
+Revision ID: 001a_pgvector_dimension_3072
+Revises: 0019_single_workspace_owner
+Create Date: 2026-08-05
+"""
+
+from __future__ import annotations
+import sqlalchemy as sa
+from alembic import op
+
+revision = '001a_pgvector_dimension_3072'
+down_revision = '0019_single_workspace_owner'
+branch_labels = None
+depends_on = None
+_TABLE = 'langbot_vectors'
+_CHECK = 'ck_langbot_vectors_embedding_dimension_enabled'
+_INDEX = 'ix_langbot_vectors_hnsw_cosine_3072'
+
+
+def upgrade() -> None:
+    conn = op.get_bind()
+    if conn.dialect.name != 'postgresql' or _TABLE not in sa.inspect(conn).get_table_names():
+        return
+    op.drop_constraint(_CHECK, _TABLE, type_='check')
+    op.create_check_constraint(_CHECK, _TABLE, 'embedding_dimension IN (384, 512, 768, 1024, 1536, 3072)')
+    op.execute(
+        sa.text(
+            f'CREATE INDEX {_INDEX} ON {_TABLE} USING hnsw ((embedding::halfvec(3072)) halfvec_cosine_ops) WHERE embedding_dimension = 3072'
+        )
+    )
+
+
+def downgrade() -> None:
+    conn = op.get_bind()
+    if conn.dialect.name != 'postgresql' or _TABLE not in sa.inspect(conn).get_table_names():
+        return
+    count = conn.scalar(sa.text(f'SELECT COUNT(*) FROM {_TABLE} WHERE embedding_dimension = 3072'))
+    if count:
+        raise RuntimeError('Cannot disable 3072-dimensional pgvector while matching embeddings exist')
+    op.drop_index(_INDEX, table_name=_TABLE)
+    op.drop_constraint(_CHECK, _TABLE, type_='check')
+    op.create_check_constraint(_CHECK, _TABLE, 'embedding_dimension IN (384, 512, 768, 1024, 1536)')

src/langbot/pkg/persistence/alembic/versions/0020_workspace_membership_source.py

@@ -0,0 +1,49 @@
+"""add explicit Workspace membership source
+
+Revision ID: 0020_membership_source
+Revises: 001a_pgvector_dimension_3072
+Create Date: 2026-08-06
+"""
+
+from __future__ import annotations
+
+import sqlalchemy as sa
+from alembic import op
+
+revision = '0020_membership_source'
+down_revision = '001a_pgvector_dimension_3072'
+branch_labels = None
+depends_on = None
+
+_CONSTRAINT_NAME = 'ck_workspace_memberships_source'
+
+
+def upgrade() -> None:
+    conn = op.get_bind()
+    inspector = sa.inspect(conn)
+    if 'workspace_memberships' not in inspector.get_table_names():
+        return
+    if 'source' in {column['name'] for column in inspector.get_columns('workspace_memberships')}:
+        return
+
+    # No durable historical field distinguishes Directory-created revision-zero
+    # rows from Core invitations. Protect every existing row; production can
+    # reclassify separately after UUIDs have been verified against Space.
+    with op.batch_alter_table('workspace_memberships') as batch_op:
+        batch_op.add_column(sa.Column('source', sa.String(length=32), nullable=False, server_default='local'))
+        batch_op.create_check_constraint(
+            _CONSTRAINT_NAME,
+            "source IN ('local', 'cloud_projection')",
+        )
+
+
+def downgrade() -> None:
+    conn = op.get_bind()
+    inspector = sa.inspect(conn)
+    if 'workspace_memberships' not in inspector.get_table_names():
+        return
+    if 'source' not in {column['name'] for column in inspector.get_columns('workspace_memberships')}:
+        return
+    with op.batch_alter_table('workspace_memberships') as batch_op:
+        batch_op.drop_constraint(_CONSTRAINT_NAME, type_='check')
+        batch_op.drop_column('source')

src/langbot/pkg/persistence/mgr.py

@@ -54,6 +54,7 @@ _ALEMBIC_TENANT_TABLES = {
     'workspace_memberships',
     'workspace_invitations',
     'workspace_execution_states',
+    'support_admin_temporary_sessions',
     'workspace_metadata',
     'api_keys',
     'bots',
@@ -97,7 +98,7 @@ _WORKSPACE_ALEMBIC_REVISION = '0009_workspace_tenancy'
 _RESOURCE_SCOPE_ALEMBIC_REVISION = '0010_scope_resources'
 _OSS_WORKSPACE_METADATA_KEY = 'oss_workspace_uuid'
 _RELEASE_MIGRATION_ADVISORY_LOCK_ID = 0x4C414E47424F5432
-_PGVECTOR_ALLOWED_DIMENSIONS = (384, 512, 768, 1024, 1536)
+_PGVECTOR_ALLOWED_DIMENSIONS = (384, 512, 768, 1024, 1536, 3072)
 _RUNTIME_SCHEMA = 'public'
 _ALEMBIC_RUNTIME_TABLE = 'alembic_version'
 _RUNTIME_TABLE_PRIVILEGES = frozenset({'SELECT', 'INSERT', 'UPDATE', 'DELETE'})
@@ -1355,14 +1356,16 @@ class PersistenceManager:
             index = by_index.get(index_name)
             index_definition = normalized(None if index is None else index['definition'])
             predicate = normalized(None if index is None else index['predicate'])
+            vector_type = 'halfvec' if dimension > 2000 else 'vector'
+            operator_class = f'{vector_type}_cosine_ops'
             if (
                 index is None
                 or index['access_method'] != 'hnsw'
                 or index['is_valid'] is not True
                 or index['is_ready'] is not True
-                or f'vector({dimension})' not in index_definition
-                or f'(embedding)::vector({dimension})' not in index_definition
-                or 'vector_cosine_ops' not in index_definition
+                or f'{vector_type}({dimension})' not in index_definition
+                or f'(embedding)::{vector_type}({dimension})' not in index_definition
+                or operator_class not in index_definition
                 or predicate.strip('() ') != f'embedding_dimension = {dimension}'
             ):
                 raise RuntimeError(f'PostgreSQL pgvector ANN index {index_name!r} is invalid')

src/langbot/pkg/persistence/tenant_uow.py

@@ -13,7 +13,7 @@ import typing
 import sqlalchemy
 import sqlalchemy.ext.asyncio as sqlalchemy_asyncio
 import sqlalchemy.orm as sqlalchemy_orm
-from pgvector.sqlalchemy import Vector
+from pgvector.sqlalchemy import HALFVEC, Vector
 from sqlalchemy.dialects.postgresql.dml import OnConflictDoNothing as PostgreSQLOnConflictDoNothing
 from sqlalchemy.dialects.postgresql.dml import OnConflictDoUpdate as PostgreSQLOnConflictDoUpdate
 from sqlalchemy.dialects.sqlite.dml import OnConflictDoNothing as SQLiteOnConflictDoNothing
@@ -43,6 +43,7 @@ TENANT_TABLE_COLUMNS: dict[str, str] = {
     'workspace_memberships': 'workspace_uuid',
     'workspace_invitations': 'workspace_uuid',
     'workspace_execution_states': 'workspace_uuid',
+    'support_admin_temporary_sessions': 'workspace_uuid',
     'workspace_metadata': 'workspace_uuid',
     'api_keys': 'workspace_uuid',
     'bots': 'workspace_uuid',
@@ -75,7 +76,6 @@ TENANT_TABLE_COLUMNS: dict[str, str] = {
 DIRECTORY_PROJECTION_TABLE_COLUMNS: dict[str, str] = {
     'directory_projection_states': 'instance_uuid',
     'directory_projection_inbox': 'instance_uuid',
-    'space_launch_assertion_consumptions': 'instance_uuid',
 }
 
 DIRECTORY_PROJECTED_TENANT_TABLES = frozenset(
@@ -209,7 +209,7 @@ _ALLOWED_SCOPED_BUILTIN_FUNCTION_TYPES = {
     'now': sqlalchemy.sql.functions.now,
     'sum': sqlalchemy.sql.functions.sum,
 }
-_ALLOWED_SCOPED_GENERIC_FUNCTIONS = frozenset({'length', 'nullif'})
+_ALLOWED_SCOPED_GENERIC_FUNCTIONS = frozenset({'date_trunc', 'length', 'nullif'})
 _ALLOWED_SCOPED_CUSTOM_OPERATORS = frozenset({'<=>'})
 _ALLOWED_SCOPED_STATEMENT_TYPES = (
     sqlalchemy.sql.dml.UpdateBase,
@@ -281,7 +281,7 @@ def _validate_scoped_sql_type(
         return
     seen.add(identity)
 
-    if type(sql_type) is Vector:
+    if type(sql_type) in {Vector, HALFVEC}:
         return
     if not type(sql_type).__module__.startswith('sqlalchemy.'):
         raise ScopedSessionTransactionError('TenantUnitOfWork does not allow custom SQL types in public statements')
@@ -462,7 +462,7 @@ def _validate_scoped_statement_call(args: tuple[typing.Any, ...], kwargs: dict[s
         if isinstance(element, sqlalchemy.sql.elements.BindParameter) and element.literal_execute:
             raise ScopedSessionTransactionError('TenantUnitOfWork does not allow literal-execute SQL parameters')
 
-        if isinstance(element, sqlalchemy.sql.elements.Cast) and type(element.type) is not Vector:
+        if isinstance(element, sqlalchemy.sql.elements.Cast) and type(element.type) not in {Vector, HALFVEC}:
             raise ScopedSessionTransactionError(
                 'TenantUnitOfWork only allows the trusted pgvector cast used by tenant vector search'
             )

src/langbot/pkg/pipeline/controller.py

@@ -132,9 +132,7 @@ class Controller:
 
                             break
 
-                    if selected_query:  # 找到了
-                        queries.remove(selected_query)
-                    else:  # 没找到 说明：没有请求 或者 所有query对应的session都已达到并发上限
+                    if not selected_query:  # 没找到 说明：没有请求 或者 所有query对应的session都已达到并发上限
                         await self.ap.query_pool.condition.wait()
                         continue
 

src/langbot/pkg/pipeline/process/handlers/chat.py

@@ -15,6 +15,7 @@ from ....provider import runner as runner_module
 import langbot_plugin.api.entities.events as events
 from ....utils import importutil, constants, runner as runner_utils
 from ....telemetry import features as telemetry_features
+from ....telemetry.identity import workspace_identity
 from ....provider import runners
 import langbot_plugin.api.entities.builtin.provider.session as provider_session
 import langbot_plugin.api.entities.builtin.pipeline.query as pipeline_query
@@ -265,7 +266,8 @@ class ChatMessageHandler(handler.MessageHandler):
                         'duration_ms': duration_ms,
                         'model_name': model_name,
                         'version': constants.semantic_version,
-                        'instance_id': constants.instance_id,
+                        **workspace_identity(get_query_execution_context(query)),
+                        'runtime_instance_id': constants.instance_id,
                         'edition': constants.edition,
                         'pipeline_plugins': pipeline_plugins,
                         'features': features,

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

@@ -179,10 +179,13 @@ class TelegramMessageConverter(abstract_platform_adapter.AbstractMessageConverte
                 )
                 file_format = 'image/jpeg'
 
+            # NOTE: Telegram's file.file_path is a full URL of the form
+            # https://api.telegram.org/file/bot<TOKEN>/<path> which embeds the
+            # bot token. Unlike the public CDN URLs used by other adapters, it
+            # cannot be exposed safely, so only base64 is stored here.
             encoded = await asyncio.to_thread(base64.b64encode, file_bytes)
             message_components.append(
                 platform_message.Image(
-                    url=file.file_path,
                     base64=f'data:{file_format};base64,{encoded.decode("utf-8")}',
                 )
             )

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

@@ -707,28 +707,37 @@ class WebSocketAdapter(abstract_platform_adapter.AbstractMessagePlatformAdapter)
             if len(listener_tasks) >= 100:
                 await self.logger.warning('WebSocket inbound listener capacity reached; dropping message')
                 return
-            token = _current_pipeline_uuid.set(pipeline_uuid)
-            try:
-                task_manager = getattr(self.ap, 'task_mgr', None)
-                if task_manager is None or not isinstance(getattr(task_manager, 'tasks', None), list):
-                    listener_task = asyncio.create_task(listeners[event.__class__](event, callback_adapter))
-                else:
-                    listener_task = task_manager.create_task(
-                        listeners[event.__class__](event, callback_adapter),
-                        kind='websocket-message',
-                        name=f'websocket-message-{connection.connection_id}',
-                        scopes=[
-                            core_entities.LifecycleControlScope.APPLICATION,
-                            core_entities.LifecycleControlScope.PLATFORM,
-                        ],
-                        instance_uuid=connection.instance_uuid,
-                        workspace_uuid=connection.workspace_uuid,
-                        placement_generation=connection.placement_generation,
-                    ).task
-                listener_tasks.add(listener_task)
-                listener_task.add_done_callback(self._listener_task_done)
-            finally:
-                _current_pipeline_uuid.reset(token)
+            listener = typing.cast(
+                typing.Callable[[typing.Any, typing.Any], typing.Awaitable[None]],
+                listeners[event.__class__],
+            )
+
+            async def run_listener():
+                token = _current_pipeline_uuid.set(pipeline_uuid)
+                try:
+                    await listener(event, callback_adapter)
+                finally:
+                    _current_pipeline_uuid.reset(token)
+
+            listener_coro = run_listener()
+            task_manager = getattr(self.ap, 'task_mgr', None)
+            if task_manager is None or not isinstance(getattr(task_manager, 'tasks', None), list):
+                listener_task = asyncio.create_task(listener_coro)
+            else:
+                listener_task = task_manager.create_task(
+                    listener_coro,
+                    kind='websocket-message',
+                    name=f'websocket-message-{connection.connection_id}',
+                    scopes=[
+                        core_entities.LifecycleControlScope.APPLICATION,
+                        core_entities.LifecycleControlScope.PLATFORM,
+                    ],
+                    instance_uuid=connection.instance_uuid,
+                    workspace_uuid=connection.workspace_uuid,
+                    placement_generation=connection.placement_generation,
+                ).task
+            listener_tasks.add(listener_task)
+            listener_task.add_done_callback(self._listener_task_done)
 
     def get_websocket_messages(
         self,

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

@@ -9,7 +9,7 @@ from datetime import datetime
 
 import pydantic
 
-from ...api.http.context import ExecutionContext
+from ...api.http.context import ExecutionContext, PrincipalContext
 
 logger = logging.getLogger(__name__)
 _SESSION_FILTER_UNSET = object()
@@ -95,6 +95,9 @@ class WebSocketConnection(pydantic.BaseModel):
     metadata: dict = pydantic.Field(default_factory=dict)
     """连接元数据（可存储额外信息）"""
 
+    trigger_principal: PrincipalContext | None = None
+    """Authenticated principal that opened this dashboard connection."""
+
     @property
     def scope(self) -> WebSocketScope:
         return WebSocketScope(
@@ -112,6 +115,7 @@ class WebSocketConnection(pydantic.BaseModel):
             workspace_uuid=self.workspace_uuid,
             placement_generation=self.placement_generation,
             pipeline_uuid=self.pipeline_uuid,
+            trigger_principal=self.trigger_principal,
         )
 
 
@@ -138,6 +142,7 @@ class WebSocketConnectionManager:
         pipeline_uuid: str,
         session_type: str,
         metadata: dict | None = None,
+        trigger_principal: PrincipalContext | None = None,
         session_id: str | None = None,
         send_queue_size: int = _DEFAULT_SEND_QUEUE_SIZE,
         max_connections: int = 1024,
@@ -174,6 +179,7 @@ class WebSocketConnectionManager:
                 session_id=session_id,
                 websocket=websocket,
                 metadata=metadata or {},
+                trigger_principal=trigger_principal,
                 send_queue=asyncio.Queue(maxsize=send_queue_size),
             )
 

src/langbot/pkg/plugin/connector.py

@@ -6,6 +6,7 @@ import contextlib
 import contextvars
 import hashlib
 import json
+import math
 import time
 import uuid
 from typing import Any
@@ -19,11 +20,6 @@ from urllib.parse import urljoin, urlparse
 from langbot_plugin.api.entities.builtin.pipeline.query import provider_session
 
 from ..core import app
-from ..cloud.quotas import (
-    lock_workspace_for_quota,
-    require_resource_capacity,
-    resolve_workspace_quota,
-)
 from . import handler
 from .archive import inspect_plugin_archive_metadata
 from .github import (
@@ -81,7 +77,7 @@ _GITHUB_ASSET_HOSTS = frozenset(
     }
 )
 _HTTP_REDIRECT_STATUSES = frozenset({301, 302, 303, 307, 308})
-_CONNECT_TIMEOUT_SEC = 30.0
+_DEFAULT_CONNECT_TIMEOUT_SECONDS = 180.0
 _HEARTBEAT_INTERVAL_SEC = 20.0
 _HEARTBEAT_FAILURE_THRESHOLD = 3
 _RECONNECT_MAX_DELAY_SEC = 60.0
@@ -211,6 +207,17 @@ class PluginRuntimeConnector(ManagedRuntimeConnector):
 
         return f'{constants.instance_id}:plugin-runtime'
 
+    @staticmethod
+    def _runtime_connect_timeout(plugin_config: dict[str, Any]) -> float:
+        value = plugin_config.get('connect_timeout_seconds', _DEFAULT_CONNECT_TIMEOUT_SECONDS)
+        if isinstance(value, bool) or not isinstance(value, (int, float)) or not math.isfinite(value) or value <= 0:
+            raise ValueError('plugin.connect_timeout_seconds must be a positive number')
+        return float(value)
+
+    @staticmethod
+    def _runtime_connect_timeout_error(timeout_seconds: float) -> str:
+        return f'Plugin runtime did not become ready within {timeout_seconds:g} seconds'
+
     def _runtime_handler(self) -> handler.RuntimeConnectionHandler:
         runtime_handler = getattr(self, 'handler', None)
         if runtime_handler is None:
@@ -256,6 +263,8 @@ class PluginRuntimeConnector(ManagedRuntimeConnector):
     def _control_headers(self, *, allow_generate: bool) -> dict[str, str]:
         if not self._control_token and allow_generate:
             self._control_token = secrets.token_urlsafe(48)
+        if not self._control_token:
+            return {}
         try:
             self._control_token = validate_runtime_secret(
                 self._control_token,
@@ -704,10 +713,13 @@ class PluginRuntimeConnector(ManagedRuntimeConnector):
         """
 
         runtime_handler = self._runtime_handler()
+        started_at = time.monotonic()
         async with self._state_lock:
             all_states: dict[str, PluginInstallationDesiredState] = {}
             workspace_installations: dict[str, set[str]] = {}
+            workspace_count = 0
             for context in contexts:
+                workspace_count += 1
                 execution_context = await self._validate_execution_context(context)
                 states = await self._load_workspace_desired_states(execution_context)
                 installation_ids = {state.binding.installation_uuid for state in states}
@@ -727,6 +739,13 @@ class PluginRuntimeConnector(ManagedRuntimeConnector):
                     runtime_handler.unregister_installation_binding(previous.binding)
             self._known_desired_states = all_states
             self._workspace_installations = workspace_installations
+            self.ap.logger.info(
+                'Shared plugin runtime reconcile completed: workspaces=%d desired_installations=%d '
+                'elapsed_seconds=%.3f',
+                workspace_count,
+                len(all_states),
+                time.monotonic() - started_at,
+            )
             return result
 
     async def _validate_execution_context(self, context: TenantContext) -> ExecutionContext:
@@ -962,11 +981,14 @@ class PluginRuntimeConnector(ManagedRuntimeConnector):
                 task_coro = self.ctrl.run(new_connection_callback)
 
             self._transport_task = asyncio.create_task(task_coro)
+            connect_timeout_seconds = self._runtime_connect_timeout(self.ap.instance_config.data.get('plugin', {}))
             try:
-                await asyncio.wait_for(self._connected.wait(), timeout=_CONNECT_TIMEOUT_SEC)
+                await asyncio.wait_for(self._connected.wait(), timeout=connect_timeout_seconds)
             except asyncio.TimeoutError as exc:
                 await self._stop_transport()
-                raise PluginRuntimeNotConnectedError('Plugin runtime did not become ready within 30 seconds') from exc
+                raise PluginRuntimeNotConnectedError(
+                    self._runtime_connect_timeout_error(connect_timeout_seconds)
+                ) from exc
             if connect_errors:
                 await self._stop_transport()
                 raise PluginRuntimeNotConnectedError(f'Plugin runtime connection failed: {connect_errors[-1]}')
@@ -1300,11 +1322,6 @@ class PluginRuntimeConnector(ManagedRuntimeConnector):
         install_info: dict[str, Any],
         artifact_digest: str,
     ) -> tuple[InstallationBinding, str | None, bool]:
-        quota = await resolve_workspace_quota(
-            self.ap,
-            execution_context.workspace_uuid,
-            'plugins.max',
-        )
         safe_install_info = {
             key: value
             for key, value in install_info.items()
@@ -1326,19 +1343,9 @@ class PluginRuntimeConnector(ManagedRuntimeConnector):
         )
 
         async def persist(execute):
-            if quota.requires_transaction_lock:
-                await lock_workspace_for_quota(execute, execution_context.workspace_uuid)
             result = await execute(statement)
             setting = result.first()
             if setting is None:
-                await require_resource_capacity(
-                    execute,
-                    workspace_uuid=execution_context.workspace_uuid,
-                    model=persistence_plugin.PluginSetting,
-                    quota=quota,
-                    resource_name='plugins',
-                    workspace_locked=quota.requires_transaction_lock,
-                )
                 installation_uuid = str(uuid.uuid4())
                 runtime_revision = 1
                 previous_digest = None
@@ -1393,8 +1400,6 @@ class PluginRuntimeConnector(ManagedRuntimeConnector):
             )
 
         tenant_uow = getattr(self.ap.persistence_mgr, 'tenant_uow', None)
-        if quota.requires_transaction_lock and not callable(tenant_uow):
-            raise RuntimeError('Cloud plugin quota enforcement requires transactional persistence')
         if callable(tenant_uow):
             async with tenant_uow(execution_context.workspace_uuid) as uow:
                 return await persist(uow.execute)
@@ -1990,11 +1995,11 @@ class PluginRuntimeConnector(ManagedRuntimeConnector):
         with runtime_handler.installation_scope(binding):
             return await runtime_handler.handle_page_api(plugin_author, plugin_name, page_id, endpoint, method, body)
 
-    async def get_debug_info(self) -> dict[str, Any]:
+    async def get_debug_info(self, execution_context: ExecutionContext) -> dict[str, Any]:
         """Get debug information including debug key and WS URL"""
         if not self.is_enable_plugin or not self._runtime_available():
             return {}
-        return await self._runtime_handler().get_debug_info()
+        return await self._runtime_handler().get_debug_info(execution_context)
 
     async def emit_event(
         self,

src/langbot/pkg/plugin/handler.py

@@ -1960,14 +1960,14 @@ class RuntimeConnectionHandler(handler.Handler):
         )
         return result
 
-    async def get_debug_info(self) -> dict[str, Any]:
+    async def get_debug_info(self, execution_context: ExecutionContext) -> dict[str, Any]:
         """Get debug information including debug key and WS URL"""
-        with self.installation_scope(None):
-            result = await self.call_action(
-                LangBotToRuntimeAction.GET_DEBUG_INFO,
-                {},
-                timeout=10,
-            )
+        result = await self.call_action(
+            LangBotToRuntimeAction.GET_DEBUG_INFO,
+            {},
+            timeout=10,
+            action_context=execution_context,
+        )
         return result
 
     # ================= RAG Capability Callers (LangBot -> Runtime) =================

src/langbot/pkg/telemetry/heartbeat.py

@@ -27,6 +27,19 @@ if typing.TYPE_CHECKING:
 HEARTBEAT_INTERVAL_SECONDS = 24 * 3600
 
 
+class WorkspaceResourceSnapshot(typing.TypedDict):
+    workspace_uuid: str
+    bot_count: int
+    pipeline_count: int
+    knowledge_base_count: int
+    plugin_count: int
+    mcp_server_count: int
+    extension_count: int
+    skill_count: int
+    adapters: list[str]
+    execution_generation: int
+
+
 async def _count(
     ap: core_app.Application,
     table,
@@ -52,14 +65,13 @@ async def _count(
         return -1
 
 
-async def _cloud_workspace_resource_counts(ap: core_app.Application) -> list[dict]:
+async def _cloud_workspace_resource_counts(ap: core_app.Application, bindings) -> list[WorkspaceResourceSnapshot]:
     """Summarize already-loaded Cloud registries without per-tenant SQL."""
     persistence_mgr = ap.persistence_mgr
     if getattr(getattr(persistence_mgr, 'mode', None), 'value', None) != 'cloud_runtime':
         return []
 
-    bindings = await ap.workspace_service.list_active_execution_bindings()
-    counts = {
+    counts: dict[str, WorkspaceResourceSnapshot] = {
         binding.workspace_uuid: {
             'workspace_uuid': binding.workspace_uuid,
             'bot_count': 0,
@@ -68,13 +80,20 @@ async def _cloud_workspace_resource_counts(ap: core_app.Application) -> list[dic
             'plugin_count': 0,
             'mcp_server_count': 0,
             'extension_count': 0,
+            'skill_count': 0,
+            'adapters': [],
+            'execution_generation': binding.placement_generation,
         }
         for binding in bindings
     }
 
-    for key in getattr(ap.platform_mgr, '_bots_by_key', {}):
+    adapter_sets: dict[str, set[str]] = {workspace_uuid: set() for workspace_uuid in counts}
+    for key, bot in getattr(ap.platform_mgr, '_bots_by_key', {}).items():
         if len(key) >= 2 and key[1] in counts:
             counts[key[1]]['bot_count'] += 1
+            adapter = getattr(bot, 'adapter', None)
+            if adapter is not None and getattr(bot, 'enable', False):
+                adapter_sets[key[1]].add(adapter.__class__.__name__)
     for key in getattr(ap.pipeline_mgr, '_pipelines_by_key', {}):
         if len(key) >= 2 and key[1] in counts:
             counts[key[1]]['pipeline_count'] += 1
@@ -87,14 +106,24 @@ async def _cloud_workspace_resource_counts(ap: core_app.Application) -> list[dic
     for workspace_uuid, installations in getattr(ap.plugin_connector, '_workspace_installations', {}).items():
         if workspace_uuid in counts:
             counts[workspace_uuid]['plugin_count'] = len(installations)
+    for key, skills in getattr(ap.skill_mgr, '_skills_by_scope', {}).items():
+        if len(key) >= 2 and key[1] in counts:
+            counts[key[1]]['skill_count'] += len(skills)
 
-    for resource in counts.values():
+    for workspace_uuid, resource in counts.items():
         resource['extension_count'] = resource['plugin_count'] + resource['mcp_server_count']
+        resource['adapters'] = sorted(adapter_sets[workspace_uuid])
     return list(counts.values())
 
 
-async def build_heartbeat_payload(ap: core_app.Application) -> dict:
-    """Collect the anonymous instance profile snapshot."""
+async def build_heartbeat_payload(
+    ap: core_app.Application,
+    *,
+    workspace_uuid: str,
+    workspace_create_ts: int = 0,
+    workspace_resource: WorkspaceResourceSnapshot | None = None,
+) -> dict:
+    """Collect one anonymous Workspace profile snapshot."""
     from ..entity.persistence import bot as persistence_bot
     from ..entity.persistence import mcp as persistence_mcp
     from ..entity.persistence import pipeline as persistence_pipeline
@@ -177,15 +206,16 @@ async def build_heartbeat_payload(ap: core_app.Application) -> dict:
     except Exception:
         pass
 
-    workspace_resources = await _cloud_workspace_resource_counts(ap)
-    if workspace_resources:
-        features['workspace_resources'] = workspace_resources
+    if workspace_resource is not None:
+        features.update({key: value for key, value in workspace_resource.items() if key != 'workspace_uuid'})
 
     return {
         'event_type': 'instance_heartbeat',
         'query_id': '',
         'version': constants.semantic_version,
         'instance_id': constants.instance_id,
+        'workspace_uuid': workspace_uuid,
+        'workspace_create_ts': workspace_create_ts,
         'instance_create_ts': constants.instance_create_ts,
         'edition': constants.edition,
         'features': features,
@@ -193,14 +223,49 @@ async def build_heartbeat_payload(ap: core_app.Application) -> dict:
     }
 
 
+def _workspace_created_timestamp(created_at: datetime | None) -> int:
+    if created_at is None:
+        return 0
+    if created_at.tzinfo is None:
+        # SQLAlchemy may return persisted UTC values without tzinfo. Never
+        # reinterpret them in the host's local timezone.
+        created_at = created_at.replace(tzinfo=timezone.utc)
+    return int(created_at.timestamp())
+
+
+async def build_heartbeat_payloads(ap: core_app.Application) -> list[dict]:
+    """Build one heartbeat per active Workspace."""
+    bindings = await ap.workspace_service.list_active_execution_bindings()
+    workspace_uuids = sorted({binding.workspace_uuid for binding in bindings})
+    workspace_create_ts = {
+        binding.workspace_uuid: _workspace_created_timestamp(getattr(binding, 'workspace_created_at', None))
+        for binding in bindings
+    }
+    resources = {
+        resource['workspace_uuid']: resource for resource in await _cloud_workspace_resource_counts(ap, bindings)
+    }
+    return [
+        await build_heartbeat_payload(
+            ap,
+            workspace_uuid=workspace_uuid,
+            workspace_create_ts=workspace_create_ts.get(workspace_uuid, 0),
+            workspace_resource=resources.get(workspace_uuid),
+        )
+        for workspace_uuid in workspace_uuids
+    ]
+
+
 async def heartbeat_loop(ap: core_app.Application) -> None:
     """Send one heartbeat shortly after startup, then daily."""
     # Small delay so managers (platform, skills, plugins) finish loading first
     await asyncio.sleep(30)
     while True:
         try:
-            payload = await build_heartbeat_payload(ap)
-            await ap.telemetry.start_send_task(payload)
+            for payload in await build_heartbeat_payloads(ap):
+                # Heartbeats are a daily bounded batch, not best-effort query events.
+                # Await each send so the TelemetryManager's 8-task queue cannot drop
+                # Workspaces after the first batch.
+                await ap.telemetry.send(payload)
         except Exception as e:
             try:
                 ap.logger.debug(f'Telemetry heartbeat failed: {e}')

src/langbot/pkg/telemetry/identity.py

@@ -0,0 +1,22 @@
+from __future__ import annotations
+
+import typing
+
+
+class WorkspaceExecutionContext(typing.Protocol):
+    @property
+    def instance_uuid(self) -> str: ...
+
+    @property
+    def workspace_uuid(self) -> str: ...
+
+
+def workspace_identity(execution_context: WorkspaceExecutionContext) -> dict[str, str]:
+    """Build both first-class telemetry identities for one execution."""
+    instance_id = execution_context.instance_uuid.strip()
+    workspace_uuid = execution_context.workspace_uuid.strip()
+    if not instance_id:
+        raise ValueError('Telemetry execution instance ID is empty')
+    if not workspace_uuid:
+        raise ValueError('Telemetry execution Workspace UUID is empty')
+    return {'instance_id': instance_id, 'workspace_uuid': workspace_uuid}

src/langbot/pkg/telemetry/telemetry.py

@@ -2,7 +2,11 @@ from __future__ import annotations
 
 import asyncio
 import contextlib
+import os
+import typing
+
 import httpx
+
 from ..core import app as core_app
 from ..utils import httpclient
 
@@ -21,7 +25,7 @@ class TelemetryManager:
     def __init__(self, ap: core_app.Application):
         self.ap = ap
 
-        self.telemetry_config = {}
+        self.telemetry_config: dict[str, typing.Any] = {}
         self.send_tasks: list[asyncio.Task] = []
         self._client: httpx.AsyncClient | None = None
 
@@ -131,7 +135,35 @@ class TelemetryManager:
                 async with self._client_context() as client:
                     try:
                         # Use asyncio.wait_for to ensure we always bound the total time
-                        resp = await asyncio.wait_for(client.post(url, json=sanitized), timeout=10 + 1)
+                        telemetry_token = os.getenv('LANGBOT_TELEMETRY_INGEST_TOKEN', '').strip()
+                        headers: dict[str, str] = {}
+                        if telemetry_token:
+                            headers['X-LangBot-Telemetry-Token'] = telemetry_token
+                        else:
+                            workspace_uuid = str(sanitized.get('workspace_uuid', '')).strip()
+                            user_service = getattr(self.ap, 'user_service', None)
+                            if workspace_uuid and user_service is not None:
+                                try:
+                                    owner = await user_service.get_workspace_owner(workspace_uuid)
+                                    owner_email = str(getattr(owner, 'user', '') or '').strip()
+                                    space_service = getattr(self.ap, 'space_service', None)
+                                    access_token = (
+                                        await space_service.get_valid_access_token(owner_email)
+                                        if owner_email and space_service is not None
+                                        else None
+                                    )
+                                    access_token = str(access_token or '').strip()
+                                    if access_token:
+                                        headers['Authorization'] = f'Bearer {access_token}'
+                                except Exception:
+                                    self.ap.logger.debug(
+                                        'Could not resolve authenticated telemetry reporter', exc_info=True
+                                    )
+                        if headers:
+                            request = client.post(url, json=sanitized, headers=headers)
+                        else:
+                            request = client.post(url, json=sanitized)
+                        resp = await asyncio.wait_for(request, timeout=10 + 1)
 
                         if resp.status_code >= 400:
                             body = await httpclient.response_text(resp, max_chars=200)
@@ -143,7 +175,8 @@ class TelemetryManager:
                             app_err = False
                             try:
                                 j = await httpclient.parse_json_response(resp)
-                                if isinstance(j, dict) and j.get('code') is not None and int(j.get('code')) >= 400:
+                                app_code = j.get('code') if isinstance(j, dict) else None
+                                if app_code is not None and int(app_code) >= 400:
                                     app_err = True
                                     self.ap.logger.warning(
                                         f'Telemetry post to {url} returned application error code {j.get("code")} - {j.get("msg")}'

src/langbot/pkg/vector/mgr.py

@@ -67,7 +67,7 @@ class VectorDBManager:
                 use_business_database = pgvector_config.get('use_business_database', False)
                 allowed_dimensions = pgvector_config.get(
                     'allowed_dimensions',
-                    [384, 512, 768, 1024, 1536],
+                    [384, 512, 768, 1024, 1536, 3072],
                 )
                 common_options = {
                     'use_business_database': use_business_database,

src/langbot/pkg/vector/vdbs/pgvector_db.py

@@ -6,7 +6,7 @@ from collections.abc import AsyncIterator
 from typing import Any
 
 import sqlalchemy
-from pgvector.sqlalchemy import Vector
+from pgvector.sqlalchemy import HALFVEC, Vector
 from sqlalchemy.dialects.postgresql import insert as postgresql_insert
 from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
 from sqlalchemy.orm import declarative_base
@@ -18,7 +18,7 @@ from langbot.pkg.vector.vdb import VectorDatabase
 
 Base = declarative_base()
 
-DEFAULT_ALLOWED_DIMENSIONS = (384, 512, 768, 1024, 1536)
+DEFAULT_ALLOWED_DIMENSIONS = (384, 512, 768, 1024, 1536, 3072)
 
 # pgvector schema only stores these metadata fields.
 _PG_SUPPORTED_FIELDS = {'text', 'file_id', 'chunk_uuid'}
@@ -321,7 +321,12 @@ class PgVectorDatabase(VectorDatabase):
         if len(query_embedding) != scope.embedding_dimension:
             raise ValueError(f'Query embedding must have the selected dimension {scope.embedding_dimension}')
 
-        typed_embedding = sqlalchemy.cast(PgVectorEntry.embedding, Vector(scope.embedding_dimension))
+        typed_embedding = sqlalchemy.cast(
+            PgVectorEntry.embedding,
+            HALFVEC(scope.embedding_dimension)
+            if scope.embedding_dimension > 2000
+            else Vector(scope.embedding_dimension),
+        )
         distance = typed_embedding.cosine_distance(query_embedding)
         statement = (
             sqlalchemy.select(

src/langbot/pkg/workspace/collaboration.py

@@ -17,6 +17,7 @@ from ..entity.persistence.user import AccountStatus, User
 from ..entity.persistence.workspace import (
     InvitationStatus,
     MembershipRole,
+    MembershipSource,
     MembershipStatus,
     Workspace,
     WorkspaceInvitation,
@@ -88,6 +89,7 @@ class ResolvedWorkspaceAccess:
 @dataclasses.dataclass(frozen=True, slots=True)
 class WorkspaceMemberView:
     membership: WorkspaceMembership
+    display_name: str
     email: str
 
 
@@ -294,7 +296,7 @@ class WorkspaceCollaborationService:
         async def operation(active_session: AsyncSession) -> list[WorkspaceMemberView]:
             await self._load_actor(active_session, workspace_uuid, actor)
             statement = (
-                sqlalchemy.select(WorkspaceMembership, User.user)
+                sqlalchemy.select(WorkspaceMembership, User.user, User.normalized_email)
                 .join(User, User.uuid == WorkspaceMembership.account_uuid)
                 .where(
                     WorkspaceMembership.workspace_uuid == workspace_uuid,
@@ -304,8 +306,12 @@ class WorkspaceCollaborationService:
                 .order_by(WorkspaceMembership.created_at, WorkspaceMembership.uuid)
             )
             return [
-                WorkspaceMemberView(membership=membership, email=email)
-                for membership, email in (await active_session.execute(statement)).all()
+                WorkspaceMemberView(
+                    membership=membership,
+                    display_name=display_name,
+                    email=email,
+                )
+                for membership, display_name, email in (await active_session.execute(statement)).all()
             ]
 
         return await self._run(operation, session=session, read_only=True)
@@ -478,6 +484,7 @@ class WorkspaceCollaborationService:
                     account_uuid=account_uuid,
                     role=invitation.role,
                     status=MembershipStatus.ACTIVE.value,
+                    source=MembershipSource.LOCAL.value,
                     invited_by_account_uuid=invitation.created_by_account_uuid,
                     joined_at=now,
                     projection_revision=0,
@@ -486,6 +493,7 @@ class WorkspaceCollaborationService:
             elif membership.status != MembershipStatus.ACTIVE.value:
                 membership.role = invitation.role
                 membership.status = MembershipStatus.ACTIVE.value
+                membership.source = MembershipSource.LOCAL.value
                 membership.invited_by_account_uuid = invitation.created_by_account_uuid
                 membership.joined_at = now
 
@@ -606,6 +614,8 @@ class WorkspaceCollaborationService:
     ) -> WorkspaceMembership:
         if role not in {item.value for item in MembershipRole}:
             raise MembershipPermissionError('Unknown Workspace role')
+        if role == MembershipRole.OWNER.value:
+            raise MembershipPermissionError('Workspace ownership cannot be transferred')
 
         async def operation(active_session: AsyncSession) -> WorkspaceMembership:
             await self._require_active_workspace(active_session, workspace_uuid)
@@ -617,8 +627,8 @@ class WorkspaceCollaborationService:
                 target_account_uuid,
             )
             self._require_can_manage_target(persisted_actor, target, new_role=role)
-            if target.role == MembershipRole.OWNER.value and role != MembershipRole.OWNER.value:
-                await self._require_another_owner(active_session, workspace_uuid, target.account_uuid)
+            if target.role == MembershipRole.OWNER.value:
+                raise LastOwnerError('The Workspace owner cannot be removed or demoted')
             target.role = role
             await active_session.flush()
             return target
@@ -644,7 +654,7 @@ class WorkspaceCollaborationService:
             )
             self._require_can_manage_target(persisted_actor, target)
             if target.role == MembershipRole.OWNER.value:
-                await self._require_another_owner(active_session, workspace_uuid, target.account_uuid)
+                raise LastOwnerError('The Workspace owner cannot be removed or demoted')
             target.status = MembershipStatus.REMOVED.value
             await active_session.flush()
             return target
@@ -751,26 +761,6 @@ class WorkspaceCollaborationService:
             raise WorkspaceNotFoundError('Workspace not found')
         return persisted_actor
 
-    async def _require_another_owner(
-        self,
-        session: AsyncSession,
-        workspace_uuid: str,
-        excluded_account_uuid: str,
-    ) -> None:
-        owners = (
-            await session.scalars(
-                sqlalchemy.select(WorkspaceMembership)
-                .where(
-                    WorkspaceMembership.workspace_uuid == workspace_uuid,
-                    WorkspaceMembership.status == MembershipStatus.ACTIVE.value,
-                    WorkspaceMembership.role == MembershipRole.OWNER.value,
-                )
-                .with_for_update()
-            )
-        ).all()
-        if not any(owner.account_uuid != excluded_account_uuid for owner in owners):
-            raise LastOwnerError('The last Workspace owner cannot be removed or demoted')
-
     def _require_actor_workspace(self, actor: WorkspaceMembership, workspace_uuid: str) -> None:
         if actor.workspace_uuid != workspace_uuid or actor.status != MembershipStatus.ACTIVE.value:
             raise WorkspaceNotFoundError('Workspace not found')

src/langbot/pkg/workspace/entities.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import datetime
 from dataclasses import dataclass
 
 
@@ -12,3 +13,4 @@ class WorkspaceExecutionBinding:
     placement_generation: int
     write_fenced: bool
     state: str
+    workspace_created_at: datetime.datetime | None = None

src/langbot/pkg/workspace/identity.py

@@ -0,0 +1,20 @@
+from __future__ import annotations
+
+import uuid
+
+
+_INSTANCE_PREFIX = 'instance_'
+_WORKSPACE_IDENTITY_NAMESPACE = uuid.UUID('8ea04f29-8528-4cc3-bb28-30a838c89d76')
+
+
+def workspace_uuid_from_instance_id(instance_id: str) -> str:
+    """Return the stable OSS Workspace UUID for a persisted instance identity."""
+    value = instance_id.strip()
+    if not value:
+        raise ValueError('LangBot instance identity is empty')
+
+    candidate = value[len(_INSTANCE_PREFIX) :] if value.startswith(_INSTANCE_PREFIX) else value
+    try:
+        return str(uuid.UUID(candidate))
+    except ValueError:
+        return str(uuid.uuid5(_WORKSPACE_IDENTITY_NAMESPACE, value))

src/langbot/pkg/workspace/service.py

@@ -11,6 +11,7 @@ from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
 
 from ..entity.persistence.workspace import (
     MembershipRole,
+    MembershipSource,
     MembershipStatus,
     Workspace,
     WorkspaceExecutionSource,
@@ -30,6 +31,7 @@ from .errors import (
     WorkspaceOwnerAlreadyExistsError,
 )
 from .entities import WorkspaceExecutionBinding
+from .identity import workspace_uuid_from_instance_id
 from .policy import CloudWorkspacePolicy, SingleWorkspacePolicy
 from .repository import WorkspaceRepository
 
@@ -282,6 +284,7 @@ class WorkspaceService:
                 placement_generation=execution_state.active_generation,
                 write_fenced=execution_state.write_fenced,
                 state=execution_state.state,
+                workspace_created_at=workspace.created_at,
             )
 
         binding = await self._run(operation, session=session)
@@ -449,6 +452,7 @@ class WorkspaceService:
                 account_uuid=account_uuid,
                 role=MembershipRole.OWNER.value,
                 status=MembershipStatus.ACTIVE.value,
+                source=MembershipSource.LOCAL.value,
                 joined_at=joined_at,
                 projection_revision=0,
             )
@@ -456,6 +460,7 @@ class WorkspaceService:
         else:
             membership.role = MembershipRole.OWNER.value
             membership.status = MembershipStatus.ACTIVE.value
+            membership.source = MembershipSource.LOCAL.value
             membership.joined_at = membership.joined_at or joined_at
 
         if workspace.created_by_account_uuid is None:
@@ -497,7 +502,7 @@ class WorkspaceService:
         created_by_account_uuid: str | None = None,
     ) -> Workspace:
         return Workspace(
-            uuid=str(uuid.uuid4()),
+            uuid=workspace_uuid_from_instance_id(self.instance_uuid),
             instance_uuid=self.instance_uuid,
             name=name,
             slug=slug,

src/langbot/templates/config.yaml

@@ -201,7 +201,7 @@ vdb:
         # keep this false when deliberately using an external pgvector DB.
         use_business_database: false
         # Release migrations create one partial ANN index per enabled value.
-        allowed_dimensions: [384, 512, 768, 1024, 1536]
+        allowed_dimensions: [384, 512, 768, 1024, 1536, 3072]
         host: '127.0.0.1'
         port: 5433
         database: 'langbot'
@@ -245,6 +245,8 @@ storage:
         max_concurrency: 16
 plugin:
     enable: true
+    # Maximum time for the Runtime transport, handshake, and desired-state replay.
+    connect_timeout_seconds: 180.0
     runtime_ws_url: 'ws://langbot_plugin_runtime:5400/control/ws'
     enable_marketplace: true
     display_plugin_debug_url: 'ws://localhost:5401/plugin/debug/ws'

tests/integration/api/test_plugins_security.py

@@ -106,7 +106,12 @@ async def plugin_security_api(plugin_module):
     application.plugin_connector.require_workspace_context = AsyncMock()
     application.plugin_connector.list_plugins = AsyncMock(return_value=[raw_plugin])
     application.plugin_connector.get_plugin_info = AsyncMock(return_value=raw_plugin)
-    application.plugin_connector.get_debug_info = AsyncMock(return_value={'plugin_debug_key': 'runtime-debug-secret'})
+    application.plugin_connector.get_debug_info = AsyncMock(
+        return_value={
+            'plugin_debug_key': 'runtime-debug-secret',
+            'expires_at': '2026-08-04T12:00:00Z',
+        }
+    )
     application.plugin_connector.get_plugin_logs = AsyncMock(return_value=['private runtime line'])
     application.plugin_connector.set_plugin_config = AsyncMock()
 
@@ -232,8 +237,9 @@ async def test_debug_key_requires_resource_manage_permission(plugin_security_api
     assert (await allowed.get_json())['data'] == {
         'debug_url': 'http://localhost:5401',
         'plugin_debug_key': 'runtime-debug-secret',
+        'expires_at': '2026-08-04T12:00:00Z',
     }
-    application.plugin_connector.get_debug_info.assert_awaited_once_with()
+    application.plugin_connector.get_debug_info.assert_awaited_once()
 
 
 @pytest.mark.asyncio

tests/integration/api/test_smoke.py

@@ -9,6 +9,8 @@ Run: uv run pytest tests/integration/api/test_smoke.py -q
 
 from __future__ import annotations
 
+from types import SimpleNamespace
+
 import pytest
 from unittest.mock import MagicMock, AsyncMock, Mock
 
@@ -304,12 +306,34 @@ class TestUserInitEndpoint:
         data = await response.get_json()
         assert data['data'] == {
             'initialized': True,
+            'authenticated_invitation_acceptance_enabled': False,
             'password_login_enabled': True,
             'space_login_enabled': False,
         }
         fake_api_app.user_service.get_login_capabilities.assert_awaited_once_with()
         fake_api_app.user_service.get_first_user.assert_not_awaited()
 
+    @pytest.mark.asyncio
+    async def test_account_info_enables_authenticated_invitation_acceptance_in_cloud(
+        self, quart_test_client, fake_api_app
+    ):
+        fake_api_app.deployment = SimpleNamespace(mode='cloud')
+        fake_api_app.user_service.is_initialized.return_value = True
+        fake_api_app.user_service.get_login_capabilities = AsyncMock(
+            return_value={'password_login_enabled': True, 'space_login_enabled': True}
+        )
+
+        response = await quart_test_client.get('/api/v1/user/account-info')
+
+        assert response.status_code == 200
+        data = await response.get_json()
+        assert data['data'] == {
+            'initialized': True,
+            'authenticated_invitation_acceptance_enabled': True,
+            'password_login_enabled': False,
+            'space_login_enabled': True,
+        }
+
     @pytest.mark.asyncio
     async def test_recovery_key_resets_any_existing_account(self, quart_test_client, fake_api_app, monkeypatch):
         fake_api_app.user_service.is_initialized.return_value = True

tests/integration/api/test_support_admin_launch.py

@@ -0,0 +1,452 @@
+from __future__ import annotations
+
+import base64
+import datetime
+import json
+import logging
+import time
+import uuid
+from types import SimpleNamespace
+from unittest.mock import AsyncMock
+
+import pytest
+import sqlalchemy
+from cryptography.hazmat.primitives import serialization
+from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
+from quart import Quart
+from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
+
+from langbot.pkg.api.http.authz import Permission
+from langbot.pkg.api.http.context import PrincipalType, RequestContext
+from langbot.pkg.api.http.controller import group
+from langbot.pkg.api.http.controller.groups.pipelines.websocket_chat import WebSocketChatRouterGroup
+from langbot.pkg.api.http.controller.groups.user import UserRouterGroup
+from langbot.pkg.cloud.launch import SpaceLaunchError, SpaceLaunchService
+from langbot.pkg.cloud.support_admin import SupportAdminSessionService
+from langbot.pkg.entity.persistence.base import Base
+from langbot.pkg.entity.persistence.support_admin import SupportAdminTemporarySession
+from langbot.pkg.entity.persistence.user import User
+from langbot.pkg.entity.persistence.workspace import (
+    Workspace,
+    WorkspaceExecutionState,
+    WorkspaceMembership,
+)
+from langbot.pkg.workspace.service import WorkspaceService
+
+
+pytestmark = [pytest.mark.integration, pytest.mark.asyncio]
+
+
+INSTANCE_UUID = 'instance-support-admin'
+WORKSPACE_UUID = '10000000-0000-4000-8000-000000000001'
+OTHER_WORKSPACE_UUID = '10000000-0000-4000-8000-000000000002'
+ACTOR_ACCOUNT_UUID = '20000000-0000-4000-8000-000000000001'
+KEY_ID = 'support-admin-key-1'
+
+
+def _base64url(raw: bytes) -> str:
+    return base64.urlsafe_b64encode(raw).rstrip(b'=').decode('ascii')
+
+
+def _sign(private_key: Ed25519PrivateKey, claims: dict, *, key_id: str = KEY_ID) -> str:
+    header = {'alg': 'EdDSA', 'kid': key_id, 'typ': 'langbot-control-plane+jwt'}
+    encoded_header = _base64url(json.dumps(header, separators=(',', ':')).encode('utf-8'))
+    encoded_claims = _base64url(json.dumps(claims, separators=(',', ':')).encode('utf-8'))
+    signing_input = f'{encoded_header}.{encoded_claims}'
+    return f'{signing_input}.{_base64url(private_key.sign(signing_input.encode("ascii")))}'
+
+
+def _admin_claims(*, now: int, jti: str | None = None, workspace_uuid: str = WORKSPACE_UUID) -> dict:
+    return {
+        'iss': 'langbot-space',
+        'aud': 'langbot-cloud-runtime',
+        'sub': f'langbot-instance:{INSTANCE_UUID}',
+        'jti': jti or str(uuid.uuid4()),
+        'iat': now,
+        'nbf': now - 5,
+        'exp': now + 90,
+        'instance_uuid': INSTANCE_UUID,
+        'kind': 'workspace.support_admin_launch',
+        'payload': {
+            'workspace_uuid': workspace_uuid,
+            'launch_mode': 'support_admin',
+            'principal_type': 'support_admin',
+            'actor_account_uuid': ACTOR_ACCOUNT_UUID,
+            'effective_role': 'owner',
+        },
+    }
+
+
+@group.group_class('support_admin_probe', '/api/v1/support-admin-probe')
+class SupportAdminProbeGroup(group.RouterGroup):
+    async def initialize(self) -> None:
+        @self.route('/user-token', auth_type=group.AuthType.USER_TOKEN, permission=Permission.WORKSPACE_VIEW)
+        async def _(request_context: RequestContext) -> str:
+            return self.success(data=_context_payload(request_context))
+
+        @self.route(
+            '/member-operation',
+            auth_type=group.AuthType.USER_TOKEN,
+            permission=Permission.MEMBER_VIEW,
+        )
+        async def member_operation(request_context: RequestContext) -> str:
+            return self.success(data=_context_payload(request_context))
+
+        @self.route(
+            '/user-token-or-api-key',
+            auth_type=group.AuthType.USER_TOKEN_OR_API_KEY,
+            permission=Permission.WORKSPACE_VIEW,
+        )
+        async def _(request_context: RequestContext) -> str:
+            return self.success(data=_context_payload(request_context))
+
+
+def _context_payload(request_context: RequestContext) -> dict:
+    return {
+        'principal_type': request_context.principal.principal_type.value,
+        'actor_account_uuid': request_context.principal.actor_account_uuid,
+        'account_uuid': request_context.principal.account_uuid,
+        'role': request_context.workspace.role,
+        'membership_uuid': request_context.workspace.membership_uuid,
+        'permissions': sorted(request_context.workspace.permissions),
+    }
+
+
+class _TenantUow:
+    def __init__(self, engine):
+        self._engine = engine
+        self.session = None
+        self._transaction = None
+
+    async def __aenter__(self):
+        session_factory = async_sessionmaker(self._engine, expire_on_commit=False)
+        self.session = session_factory()
+        self._transaction = await self.session.begin()
+        return self
+
+    async def __aexit__(self, exc_type, exc, traceback):
+        try:
+            if exc_type is None:
+                await self._transaction.commit()
+            else:
+                await self._transaction.rollback()
+        finally:
+            await self.session.close()
+
+
+class _TenantScope:
+    async def __aenter__(self):
+        return self
+
+    async def __aexit__(self, exc_type, exc, traceback):
+        return False
+
+
+class _PersistenceManager:
+    def __init__(self, engine):
+        self._engine = engine
+        self.mode = SimpleNamespace(value='oss_compat')
+
+    def get_db_engine(self):
+        return self._engine
+
+    def tenant_uow(self, workspace_uuid: str):
+        del workspace_uuid
+        return _TenantUow(self._engine)
+
+    def tenant_scope(self, workspace_uuid: str):
+        del workspace_uuid
+        return _TenantScope()
+
+
+@pytest.fixture
+async def support_admin_api(tmp_path):
+    private_key = Ed25519PrivateKey.generate()
+    public_key = private_key.public_key().public_bytes(
+        encoding=serialization.Encoding.Raw,
+        format=serialization.PublicFormat.Raw,
+    )
+    engine = create_async_engine(f'sqlite+aiosqlite:///{tmp_path / "support-admin.db"}')
+    async with engine.begin() as connection:
+        await connection.run_sync(
+            Base.metadata.create_all,
+            tables=[
+                User.__table__,
+                Workspace.__table__,
+                WorkspaceExecutionState.__table__,
+                WorkspaceMembership.__table__,
+                SupportAdminTemporarySession.__table__,
+            ],
+        )
+        for workspace_uuid, slug in (
+            (WORKSPACE_UUID, 'support-admin-a'),
+            (OTHER_WORKSPACE_UUID, 'support-admin-b'),
+        ):
+            await connection.execute(
+                sqlalchemy.insert(Workspace).values(
+                    uuid=workspace_uuid,
+                    instance_uuid=INSTANCE_UUID,
+                    name=slug,
+                    slug=slug,
+                    type='team',
+                    status='active',
+                    source='cloud_projection',
+                    projection_revision=1,
+                )
+            )
+            await connection.execute(
+                sqlalchemy.insert(WorkspaceExecutionState).values(
+                    workspace_uuid=workspace_uuid,
+                    instance_uuid=INSTANCE_UUID,
+                    active_generation=1,
+                    state='active',
+                    write_fenced=False,
+                    source='cloud',
+                    desired_state_revision=1,
+                )
+            )
+
+    app = SimpleNamespace()
+    app.persistence_mgr = _PersistenceManager(engine)
+    app.instance_config = SimpleNamespace(
+        data={
+            'system': {
+                'jwt': {'secret': 'support-admin-secret', 'expire': 3600},
+                'websocket_retention': {},
+            },
+            'space': {
+                'launch': {
+                    'control_plane_public_key': _base64url(public_key),
+                }
+            },
+            'api': {'global_api_key': ''},
+        }
+    )
+    app.logger = logging.getLogger('support-admin-test')
+    app.deployment = SimpleNamespace(mode='cloud', multi_workspace_enabled=True, verification_key_id=KEY_ID)
+    app.directory_projection_service = SimpleNamespace(require_ready=lambda: None)
+    app.workspace_service = WorkspaceService(app, instance_uuid=INSTANCE_UUID)
+    app.entitlement_resolver = SimpleNamespace(
+        instance_uuid=INSTANCE_UUID,
+        resolve=AsyncMock(return_value=SimpleNamespace(entitlement_revision=7)),
+    )
+    app.support_admin_session_service = SupportAdminSessionService(app)
+    app.space_launch_service = SpaceLaunchService(app)
+    app.user_service = SimpleNamespace()
+    app.user_service.get_authenticated_account = AsyncMock(side_effect=AssertionError('normal account auth used'))
+    app.user_service.verify_jwt_token = AsyncMock(side_effect=AssertionError('normal token verification used'))
+    app.user_service.get_user_by_email = AsyncMock(side_effect=AssertionError('user lookup used'))
+    app.apikey_service = SimpleNamespace()
+    app.apikey_service.authenticate_api_key = AsyncMock(
+        return_value=SimpleNamespace(
+            instance_uuid=INSTANCE_UUID,
+            workspace_uuid=OTHER_WORKSPACE_UUID,
+            placement_generation=1,
+            api_key_uuid='api-key',
+            permissions=frozenset(permission.value for permission in Permission),
+        )
+    )
+
+    quart_app = Quart(__name__)
+    await UserRouterGroup(app, quart_app).initialize()
+    await SupportAdminProbeGroup(app, quart_app).initialize()
+
+    yield app, quart_app.test_client(), engine, private_key
+    await engine.dispose()
+
+
+async def _issue_support_token(app, private_key: Ed25519PrivateKey, *, jti: str | None = None) -> dict[str, str]:
+    launch = await app.space_launch_service.consume_assertion(
+        _sign(private_key, _admin_claims(now=int(time.time()), jti=jti)),
+        expected_workspace_uuid=WORKSPACE_UUID,
+    )
+    return launch
+
+
+def _auth(token: str, workspace_uuid: str = WORKSPACE_UUID) -> dict[str, str]:
+    return {'Authorization': f'Bearer {token}', 'X-Workspace-Id': workspace_uuid}
+
+
+async def test_support_admin_membership_only_routes_are_denied(support_admin_api):
+    app, client, _engine, private_key = support_admin_api
+    launch = await _issue_support_token(app, private_key)
+
+    response = await client.get(
+        '/api/v1/support-admin-probe/member-operation',
+        headers=_auth(launch['support_admin_token']),
+    )
+
+    assert response.status_code == 403
+    assert (await response.get_json())['code'] == 'permission_denied'
+
+
+async def test_support_admin_check_token_is_rejected(support_admin_api):
+    app, client, _engine, private_key = support_admin_api
+    launch = await _issue_support_token(app, private_key)
+
+    response = await client.get('/api/v1/user/check-token', headers=_auth(launch['support_admin_token']))
+
+    assert response.status_code == 401
+    assert (await response.get_json())['code'] == 'invalid_authentication'
+
+
+async def test_support_admin_cross_workspace_denied_for_user_token_and_or_api_key(support_admin_api):
+    app, client, _engine, private_key = support_admin_api
+    launch = await _issue_support_token(app, private_key)
+
+    missing_selector = await client.get(
+        '/api/v1/support-admin-probe/user-token',
+        headers={'Authorization': f'Bearer {launch["support_admin_token"]}'},
+    )
+    user_response = await client.get(
+        '/api/v1/support-admin-probe/user-token',
+        headers=_auth(launch['support_admin_token'], OTHER_WORKSPACE_UUID),
+    )
+    either_response = await client.get(
+        '/api/v1/support-admin-probe/user-token-or-api-key',
+        headers={
+            **_auth(launch['support_admin_token'], OTHER_WORKSPACE_UUID),
+            'X-API-Key': 'valid-api-key',
+        },
+    )
+
+    assert missing_selector.status_code == 400
+    assert user_response.status_code == 401
+    assert either_response.status_code == 401
+    app.apikey_service.authenticate_api_key.assert_not_awaited()
+
+
+async def test_support_admin_request_context_has_actor_owner_and_no_membership(support_admin_api):
+    app, client, engine, private_key = support_admin_api
+    before_count = await _membership_count(engine)
+    launch = await _issue_support_token(app, private_key)
+
+    response = await client.get(
+        '/api/v1/support-admin-probe/user-token',
+        headers=_auth(launch['support_admin_token']),
+    )
+
+    assert response.status_code == 200
+    data = (await response.get_json())['data']
+    permissions = set(data.pop('permissions'))
+    assert Permission.WORKSPACE_VIEW.value in permissions
+    assert Permission.RESOURCE_MANAGE.value in permissions
+    assert not permissions.intersection(
+        {
+            Permission.MEMBER_VIEW.value,
+            Permission.MEMBER_INVITE.value,
+            Permission.MEMBER_UPDATE_ROLE.value,
+            Permission.MEMBER_REMOVE.value,
+        }
+    )
+    assert data == {
+        'principal_type': PrincipalType.SUPPORT_ADMIN.value,
+        'actor_account_uuid': ACTOR_ACCOUNT_UUID,
+        'account_uuid': None,
+        'role': 'owner',
+        'membership_uuid': None,
+    }
+    assert await _membership_count(engine) == before_count
+
+
+async def test_support_admin_missing_workspace_is_controlled_launch_failure(support_admin_api):
+    app, _client, engine, private_key = support_admin_api
+    async with engine.begin() as connection:
+        await connection.execute(
+            sqlalchemy.delete(WorkspaceExecutionState).where(WorkspaceExecutionState.workspace_uuid == WORKSPACE_UUID)
+        )
+
+    with pytest.raises(SpaceLaunchError, match='unavailable'):
+        await _issue_support_token(app, private_key)
+
+
+async def test_support_admin_launch_replay_is_durable_across_service_instances(support_admin_api):
+    app, _client, _engine, private_key = support_admin_api
+    jti = str(uuid.uuid4())
+
+    await _issue_support_token(app, private_key, jti=jti)
+    second_service = SpaceLaunchService(app)
+
+    with pytest.raises(SpaceLaunchError, match='already been consumed'):
+        await second_service.consume_assertion(
+            _sign(private_key, _admin_claims(now=int(time.time()), jti=jti)),
+            expected_workspace_uuid=WORKSPACE_UUID,
+        )
+
+
+async def test_support_admin_persisted_expiry_and_revocation_are_enforced(support_admin_api):
+    app, client, engine, private_key = support_admin_api
+    launch = await _issue_support_token(app, private_key)
+    token = launch['support_admin_token']
+
+    async with engine.begin() as connection:
+        await connection.execute(
+            sqlalchemy.update(SupportAdminTemporarySession)
+            .where(SupportAdminTemporarySession.grant_jti_hash == launch['grant_jti_hash'])
+            .values(expires_at=datetime.datetime.now(datetime.UTC).replace(tzinfo=None) - datetime.timedelta(minutes=1))
+        )
+    expired = await client.get('/api/v1/support-admin-probe/user-token', headers=_auth(token))
+    assert expired.status_code == 401
+
+    second = await _issue_support_token(app, private_key)
+    async with engine.begin() as connection:
+        await connection.execute(
+            sqlalchemy.update(SupportAdminTemporarySession)
+            .where(SupportAdminTemporarySession.grant_jti_hash == second['grant_jti_hash'])
+            .values(revoked_at=datetime.datetime.now(datetime.UTC).replace(tzinfo=None))
+        )
+    revoked = await client.get('/api/v1/support-admin-probe/user-token', headers=_auth(second['support_admin_token']))
+    assert revoked.status_code == 401
+
+
+async def test_support_admin_websocket_preserves_actor_and_revalidates(support_admin_api):
+    app, _client, _engine, private_key = support_admin_api
+    launch = await _issue_support_token(app, private_key)
+    captured_contexts = []
+
+    class Adapter:
+        async def handle_websocket_message(self, connection, data):
+            del data
+            captured_contexts.append(connection.execution_context)
+            await connection.send_queue.put({'type': 'handled'})
+            connection.is_active = False
+
+    app.pipeline_service = SimpleNamespace(get_pipeline=AsyncMock(return_value=SimpleNamespace(uuid='pipeline-1')))
+    app.platform_mgr = SimpleNamespace(
+        get_websocket_proxy_bot=AsyncMock(return_value=SimpleNamespace(adapter=Adapter()))
+    )
+
+    quart_app = Quart(__name__)
+    await WebSocketChatRouterGroup(app, quart_app).initialize()
+
+    async with quart_app.test_client().websocket('/api/v1/pipelines/pipeline-1/ws/connect') as websocket:
+        await websocket.send(
+            json.dumps(
+                {
+                    'type': 'authenticate',
+                    'token': launch['support_admin_token'],
+                    'workspace_uuid': WORKSPACE_UUID,
+                }
+            )
+        )
+        connected = json.loads(await websocket.receive())
+        assert connected['type'] == 'connected'
+        await websocket.send(json.dumps({'type': 'message', 'message': [{'type': 'text', 'text': 'hi'}]}))
+        handled = json.loads(await websocket.receive())
+        assert handled['type'] == 'handled'
+
+    assert captured_contexts
+    principal = captured_contexts[0].trigger_principal
+    assert principal is not None
+    assert principal.principal_type == PrincipalType.SUPPORT_ADMIN
+    assert principal.actor_account_uuid == ACTOR_ACCOUNT_UUID
+
+
+async def _membership_count(engine) -> int:
+    async with engine.connect() as connection:
+        return int(
+            await connection.scalar(
+                sqlalchemy.select(sqlalchemy.func.count()).select_from(WorkspaceMembership),
+            )
+            or 0
+        )

tests/integration/api/test_user_space_oauth.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import datetime
 from types import SimpleNamespace
 from unittest.mock import AsyncMock, Mock
 from urllib.parse import parse_qs, urlsplit
@@ -14,6 +15,7 @@ from langbot.pkg.api.http.controller.groups.user import UserRouterGroup
 
 pytestmark = pytest.mark.integration
 WORKSPACE_UUID = '11111111-1111-4111-8111-111111111111'
+WORKSPACE_CREATED_AT = datetime.datetime(2026, 1, 2, 3, 4, 5, tzinfo=datetime.UTC)
 
 
 @pytest.fixture
@@ -58,6 +60,12 @@ async def space_oauth_api():
         return_value={'account_uuid': 'account-a', 'workspace_uuid': WORKSPACE_UUID}
     )
     application.workspace_collaboration_service.resolve_account_workspace = AsyncMock(return_value=access)
+    application.workspace_service.get_execution_binding = AsyncMock(
+        return_value=SimpleNamespace(
+            workspace_uuid=WORKSPACE_UUID,
+            workspace_created_at=WORKSPACE_CREATED_AT,
+        )
+    )
     application.space_service.get_oauth_authorize_url = Mock(
         side_effect=lambda redirect_uri, state: f'https://space.example/authorize?state={state}'
     )
@@ -234,7 +242,11 @@ async def test_login_callback_requires_and_consumes_server_state(space_oauth_api
     assert response.status_code == 200
     assert (await response.get_json())['data']['token'] == 'space-login-token'
     application.user_service.consume_space_oauth_state_details.assert_awaited_once_with('opaque-login-state', 'login')
-    application.space_service.exchange_oauth_code.assert_awaited_once_with('oauth-code')
+    application.space_service.exchange_oauth_code.assert_awaited_once_with(
+        'oauth-code',
+        [WORKSPACE_UUID],
+        {WORKSPACE_UUID: int(WORKSPACE_CREATED_AT.timestamp())},
+    )
 
 
 @pytest.mark.asyncio
@@ -270,11 +282,12 @@ async def test_space_credits_are_resolved_from_workspace_owner(space_oauth_api):
 
     response = await client.get(
         '/api/v1/user/space-credits',
-        headers={'Authorization': 'Bearer account-token', 'X-Workspace-UUID': WORKSPACE_UUID},
+        headers={'Authorization': 'Bearer account-token', 'X-Workspace-Id': WORKSPACE_UUID},
     )
+    payload = await response.get_json()
 
     assert response.status_code == 200
-    assert (await response.get_json())['data'] == {
+    assert payload['data'] == {
         'credits': 25000,
         'owner_space_bound': True,
         'is_workspace_owner': True,
@@ -282,6 +295,31 @@ async def test_space_credits_are_resolved_from_workspace_owner(space_oauth_api):
     application.space_service.get_credits.assert_awaited_once_with('owner@example.com')
 
 
+@pytest.mark.asyncio
+async def test_cloud_workspace_owner_is_always_space_bound_after_login(space_oauth_api):
+    application, client = space_oauth_api
+    application.deployment.mode = 'cloud'
+    application.user_service.get_workspace_owner = AsyncMock(return_value=None)
+    application.space_service.get_credits = AsyncMock()
+    application.cloud_model_catalog_service = SimpleNamespace(
+        get_workspace_credits=lambda workspace_uuid: 25000 if workspace_uuid == WORKSPACE_UUID else None
+    )
+
+    response = await client.get(
+        '/api/v1/user/space-credits',
+        headers={'Authorization': 'Bearer account-token', 'X-Workspace-Id': WORKSPACE_UUID},
+    )
+    payload = await response.get_json()
+
+    assert response.status_code == 200
+    assert payload['data'] == {
+        'credits': 25000,
+        'owner_space_bound': True,
+        'is_workspace_owner': True,
+    }
+    application.space_service.get_credits.assert_not_awaited()
+
+
 @pytest.mark.asyncio
 async def test_bind_callback_uses_opaque_state_and_never_treats_it_as_jwt(space_oauth_api):
     application, client = space_oauth_api

tests/integration/api/test_workspaces.py

@@ -188,6 +188,7 @@ async def test_owner_invites_second_account_and_secret_is_not_persisted(workspac
     workspace_uuid = current['workspace']['uuid']
     assert current['membership']['role'] == 'owner'
     assert 'member.invite' in current['permissions']
+    assert 'owner.transfer' not in current['permissions']
 
     invite_response = await client.post(
         f'/api/v1/workspaces/{workspace_uuid}/invitations',
@@ -263,6 +264,14 @@ async def test_owner_invites_second_account_and_secret_is_not_persisted(workspac
     assert member_current['membership']['role'] == 'viewer'
     assert 'member.invite' not in member_current['permissions']
 
+    transfer_response = await client.patch(
+        f'/api/v1/workspaces/{workspace_uuid}/members/{member_current["membership"]["account_uuid"]}',
+        headers=_auth(owner_token, workspace_uuid),
+        json={'role': 'owner'},
+    )
+    assert transfer_response.status_code == 403
+    assert (await transfer_response.get_json())['code'] == 'permission_denied'
+
     forbidden_invite = await client.post(
         f'/api/v1/workspaces/{workspace_uuid}/invitations',
         headers=_auth(member_token, workspace_uuid),
@@ -272,6 +281,31 @@ async def test_owner_invites_second_account_and_secret_is_not_persisted(workspac
     assert (await forbidden_invite.get_json())['code'] == 'permission_denied'
 
 
+async def test_workspace_member_list_returns_display_name_and_email(workspace_api):
+    _, client, engine, owner_token = workspace_api
+
+    current_response = await client.get('/api/v1/workspaces/current', headers=_auth(owner_token))
+    current = (await current_response.get_json())['data']
+    workspace_uuid = current['workspace']['uuid']
+    owner_uuid = current['membership']['account_uuid']
+
+    async with engine.begin() as connection:
+        await connection.execute(
+            sqlalchemy.update(User).where(User.uuid == owner_uuid).values(user='Owner Display Name')
+        )
+
+    response = await client.get(
+        f'/api/v1/workspaces/{workspace_uuid}/members',
+        headers=_auth(owner_token, workspace_uuid),
+    )
+
+    assert response.status_code == 200
+    members = (await response.get_json())['data']['members']
+    assert len(members) == 1
+    assert members[0]['display_name'] == 'Owner Display Name'
+    assert members[0]['email'] == 'owner@example.com'
+
+
 async def test_oss_invitation_accept_requires_logout_before_registration(workspace_api):
     _, client, _, owner_token = workspace_api
 

tests/integration/persistence/test_membership_source_migration.py

@@ -0,0 +1,70 @@
+from __future__ import annotations
+
+import pytest
+import sqlalchemy as sa
+from sqlalchemy.ext.asyncio import create_async_engine
+
+from langbot.pkg.persistence.alembic_runner import run_alembic_stamp, run_alembic_upgrade
+
+
+@pytest.mark.asyncio
+async def test_membership_source_migration_backfills_existing_rows_as_local_and_enforces_constraint(tmp_path):
+    engine = create_async_engine(f'sqlite+aiosqlite:///{tmp_path / "membership-source.db"}')
+    try:
+        async with engine.begin() as connection:
+            await connection.execute(
+                sa.text(
+                    """
+                    CREATE TABLE workspace_memberships (
+                        uuid VARCHAR(36) PRIMARY KEY,
+                        workspace_uuid VARCHAR(36) NOT NULL,
+                        account_uuid VARCHAR(36) NOT NULL,
+                        role VARCHAR(32) NOT NULL,
+                        status VARCHAR(32) NOT NULL,
+                        projection_revision BIGINT NOT NULL DEFAULT 0,
+                        created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
+                        updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP
+                    )
+                    """
+                )
+            )
+            await connection.execute(
+                sa.text(
+                    """
+                    INSERT INTO workspace_memberships
+                        (uuid, workspace_uuid, account_uuid, role, status, projection_revision)
+                    VALUES
+                        ('00000000-0000-4000-8000-000000000001', 'workspace', 'local-account',
+                         'viewer', 'active', 0),
+                        ('00000000-0000-4000-8000-000000000002', 'workspace', 'cloud-account',
+                         'viewer', 'active', 0)
+                    """
+                )
+            )
+
+        await run_alembic_stamp(engine, '0019_single_workspace_owner')
+        await run_alembic_upgrade(engine, 'head')
+
+        async with engine.connect() as connection:
+            rows = (
+                await connection.execute(sa.text('SELECT uuid, source FROM workspace_memberships ORDER BY uuid'))
+            ).all()
+            columns = await connection.run_sync(
+                lambda sync_connection: {
+                    column['name']: column
+                    for column in sa.inspect(sync_connection).get_columns('workspace_memberships')
+                }
+            )
+        assert rows == [
+            ('00000000-0000-4000-8000-000000000001', 'local'),
+            ('00000000-0000-4000-8000-000000000002', 'local'),
+        ]
+        assert columns['source']['nullable'] is False
+
+        with pytest.raises(sa.exc.IntegrityError):
+            async with engine.begin() as connection:
+                await connection.execute(
+                    sa.text("UPDATE workspace_memberships SET source = 'guessed-from-user-source'")
+                )
+    finally:
+        await engine.dispose()

tests/integration/persistence/test_migrations.py

@@ -95,6 +95,18 @@ class TestSQLiteMigrationBaseline:
 class TestSQLiteMigrationUpgrade:
     """Tests for upgrade to head workflow."""
 
+    @pytest.mark.asyncio
+    async def test_upgrade_from_published_space_launch_head_to_merged_head(self, sqlite_engine):
+        """A database released at the production-only 0016 head must remain upgradable."""
+        async with sqlite_engine.begin() as conn:
+            await conn.run_sync(Base.metadata.create_all)
+
+        await run_alembic_stamp(sqlite_engine, '0016_space_launch_replay')
+        await run_alembic_upgrade(sqlite_engine, 'head')
+
+        assert await get_alembic_current(sqlite_engine) == _get_script_head()
+        assert _get_script_head() == '0020_membership_source'
+
     @pytest.mark.asyncio
     async def test_upgrade_from_baseline_to_head(self, sqlite_engine):
         """

tests/integration/persistence/test_pgvector_postgres.py

@@ -85,6 +85,32 @@ async def clean_database(postgres_engine: AsyncEngine):
     await clean()
 
 
+async def test_upgrade_adds_3072_dimension_index_and_constraint(
+    postgres_engine: AsyncEngine,
+    clean_database,
+) -> None:
+    async with postgres_engine.begin() as conn:
+        await conn.execute(text('CREATE EXTENSION IF NOT EXISTS vector'))
+        await conn.run_sync(Base.metadata.create_all)
+    await run_alembic_stamp(postgres_engine, '0010_scope_resources')
+    await run_alembic_upgrade(postgres_engine, 'head')
+
+    async with postgres_engine.connect() as conn:
+        constraint = await conn.scalar(
+            text(
+                'SELECT pg_get_constraintdef(oid) FROM pg_constraint '
+                "WHERE conrelid = 'langbot_vectors'::regclass "
+                "AND conname = 'ck_langbot_vectors_embedding_dimension_enabled'"
+            )
+        )
+        assert '3072' in constraint
+        index_definition = await conn.scalar(
+            text("SELECT indexdef FROM pg_indexes WHERE indexname = 'ix_langbot_vectors_hnsw_cosine_3072'")
+        )
+        assert 'halfvec(3072)' in index_definition
+        assert 'halfvec_cosine_ops' in index_definition
+
+
 async def test_legacy_upgrade_temporarily_suspends_and_restores_source_rls_for_unprivileged_owner(
     postgres_url: str,
     postgres_engine: AsyncEngine,

tests/integration/persistence/test_release_migration_postgres.py

@@ -92,7 +92,7 @@ def _application(postgres_url: str, *, runtime_role: str = 'langbot_runtime_not_
                     'use': 'pgvector',
                     'pgvector': {
                         'use_business_database': True,
-                        'allowed_dimensions': [384, 512, 768, 1024, 1536],
+                        'allowed_dimensions': [384, 512, 768, 1024, 1536, 3072],
                     },
                 },
             }

tests/integration/persistence/test_single_owner_migration.py

@@ -0,0 +1,100 @@
+from __future__ import annotations
+
+import pytest
+import sqlalchemy as sa
+from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
+
+from langbot.pkg.entity.persistence.base import Base
+from langbot.pkg.entity.persistence.user import User
+from langbot.pkg.entity.persistence.workspace import Workspace, WorkspaceMembership
+from langbot.pkg.persistence.alembic_runner import run_alembic_stamp, run_alembic_upgrade
+
+
+@pytest.mark.asyncio
+async def test_single_owner_migration_demotes_historical_extra_owner_and_installs_unique_index(tmp_path):
+    engine = create_async_engine(f'sqlite+aiosqlite:///{tmp_path / "single-owner.db"}')
+    try:
+        async with engine.begin() as connection:
+            await connection.run_sync(Base.metadata.create_all)
+            await connection.execute(sa.text('DROP INDEX uq_workspace_memberships_one_active_owner'))
+
+        session_factory = async_sessionmaker(engine, expire_on_commit=False)
+        workspace_uuid = '00000000-0000-4000-8000-000000000001'
+        creator_uuid = '00000000-0000-4000-8000-000000000010'
+        promoted_uuid = '00000000-0000-4000-8000-000000000020'
+        async with session_factory() as session:
+            session.add_all(
+                [
+                    User(
+                        uuid=creator_uuid,
+                        user='creator@example.test',
+                        normalized_email='creator@example.test',
+                        password='hash',
+                        account_type='local',
+                    ),
+                    User(
+                        uuid=promoted_uuid,
+                        user='promoted@example.test',
+                        normalized_email='promoted@example.test',
+                        password='hash',
+                        account_type='local',
+                    ),
+                    Workspace(
+                        uuid=workspace_uuid,
+                        instance_uuid='instance-test',
+                        name='Workspace',
+                        slug='workspace',
+                        type='team',
+                        status='active',
+                        source='local',
+                        created_by_account_uuid=creator_uuid,
+                    ),
+                    WorkspaceMembership(
+                        uuid='00000000-0000-4000-8000-000000000100',
+                        workspace_uuid=workspace_uuid,
+                        account_uuid=creator_uuid,
+                        role='owner',
+                        status='active',
+                    ),
+                    WorkspaceMembership(
+                        uuid='00000000-0000-4000-8000-000000000200',
+                        workspace_uuid=workspace_uuid,
+                        account_uuid=promoted_uuid,
+                        role='owner',
+                        status='active',
+                    ),
+                ]
+            )
+            await session.commit()
+
+        await run_alembic_stamp(engine, '0018_merge_launch_replay')
+        await run_alembic_upgrade(engine, 'head')
+
+        async with engine.connect() as connection:
+            roles = dict(
+                (
+                    await connection.execute(
+                        sa.text(
+                            'SELECT account_uuid, role FROM workspace_memberships '
+                            'WHERE workspace_uuid = :workspace_uuid ORDER BY account_uuid'
+                        ),
+                        {'workspace_uuid': workspace_uuid},
+                    )
+                ).all()
+            )
+            assert roles == {creator_uuid: 'owner', promoted_uuid: 'admin'}
+            indexes = await connection.run_sync(
+                lambda sync_connection: {
+                    index['name'] for index in sa.inspect(sync_connection).get_indexes('workspace_memberships')
+                }
+            )
+            assert 'uq_workspace_memberships_one_active_owner' in indexes
+
+        with pytest.raises(sa.exc.IntegrityError):
+            async with engine.begin() as connection:
+                await connection.execute(
+                    sa.text("UPDATE workspace_memberships SET role = 'owner' WHERE account_uuid = :account_uuid"),
+                    {'account_uuid': promoted_uuid},
+                )
+    finally:
+        await engine.dispose()

tests/integration/persistence/test_workspace_migration.py

@@ -22,6 +22,7 @@ from langbot.pkg.persistence.alembic_runner import (
 from langbot.pkg.utils import constants
 from langbot.pkg.utils import importutil
 from langbot.pkg.workspace.collaboration import normalize_email
+from langbot.pkg.workspace.identity import workspace_uuid_from_instance_id
 
 
 pytestmark = [pytest.mark.integration, pytest.mark.asyncio]
@@ -109,6 +110,7 @@ async def test_legacy_instance_gets_stable_accounts_and_default_workspace(legacy
             .mappings()
             .one()
         )
+        assert workspace['uuid'] == workspace_uuid_from_instance_id('instance_migration_test')
         assert workspace['instance_uuid'] == 'instance_migration_test'
         assert workspace['slug'] == 'default'
         assert workspace['status'] == 'active'
@@ -149,6 +151,53 @@ async def test_workspace_upgrade_is_idempotent_and_preserves_identifiers(legacy_
     assert workspace_uuid_after == workspace_uuid_before
 
 
+async def test_existing_oss_workspace_is_rekeyed_to_instance_identity(tmp_path):
+    engine = create_async_engine(f'sqlite+aiosqlite:///{tmp_path / "workspace-rekey.db"}')
+    instance_id = 'instance_a711d9e4-0953-443f-a0e9-7dd50193a79f'
+    old_workspace_uuid = '11111111-1111-4111-8111-111111111111'
+    canonical_uuid = workspace_uuid_from_instance_id(instance_id)
+    schema = sa.MetaData()
+    sa.Table(
+        'metadata',
+        schema,
+        sa.Column('key', sa.String(255), primary_key=True),
+        sa.Column('value', sa.String(255)),
+    )
+    sa.Table(
+        'workspaces',
+        schema,
+        sa.Column('uuid', sa.String(36), primary_key=True),
+        sa.Column('instance_uuid', sa.String(255), nullable=False),
+        sa.Column('slug', sa.String(255), nullable=False),
+        sa.Column('source', sa.String(32), nullable=False),
+    )
+    sa.Table(
+        'tenant_rows',
+        schema,
+        sa.Column('id', sa.Integer, primary_key=True),
+        sa.Column('workspace_uuid', sa.String(36), sa.ForeignKey('workspaces.uuid'), nullable=False),
+    )
+    async with engine.begin() as conn:
+        await conn.run_sync(schema.create_all)
+        await conn.execute(sa.text("INSERT INTO metadata (key, value) VALUES ('instance_uuid', :value)"), {'value': instance_id})
+        await conn.execute(
+            sa.text("INSERT INTO workspaces (uuid, instance_uuid, slug, source) VALUES (:uuid, :instance, 'default', 'local')"),
+            {'uuid': old_workspace_uuid, 'instance': instance_id},
+        )
+        await conn.execute(
+            sa.text("INSERT INTO tenant_rows (id, workspace_uuid) VALUES (1, :uuid)"),
+            {'uuid': old_workspace_uuid},
+        )
+    await run_alembic_stamp(engine, '0016_support_admin_sessions')
+
+    await run_alembic_upgrade(engine, 'head')
+
+    async with engine.connect() as conn:
+        assert (await conn.execute(sa.text("SELECT uuid FROM workspaces"))).scalar_one() == canonical_uuid
+        assert (await conn.execute(sa.text("SELECT workspace_uuid FROM tenant_rows"))).scalar_one() == canonical_uuid
+    await engine.dispose()
+
+
 async def test_workspace_kernel_upgrade_downgrade_upgrade_round_trip(tmp_path):
     engine = create_async_engine(f'sqlite+aiosqlite:///{tmp_path / "workspace-round-trip.db"}')
     try:
@@ -362,6 +411,47 @@ async def test_persistence_startup_defers_workspace_tables_until_account_upgrade
         await engine.dispose()
 
 
+async def test_oss_workspace_identity_rekeys_fk_graph_and_metadata(tmp_path):
+    engine = create_async_engine(f'sqlite+aiosqlite:///{tmp_path / "workspace-rekey.db"}')
+    try:
+        await _create_legacy_schema(engine)
+        await run_alembic_stamp(engine, '0008_mcp_resource_prefs')
+        await run_alembic_upgrade(engine, '0016_support_admin_sessions')
+
+        async with engine.begin() as conn:
+            old_uuid = await conn.scalar(sa.text("SELECT uuid FROM workspaces WHERE source = 'local'"))
+            instance_uuid = await conn.scalar(sa.text("SELECT instance_uuid FROM workspaces WHERE source = 'local'"))
+            assert old_uuid
+            assert instance_uuid
+            await conn.execute(
+                sa.text(
+                    "INSERT INTO workspace_metadata (workspace_uuid, key, value) "
+                    "VALUES (:workspace_uuid, 'migration_probe', 'present')"
+                ),
+                {'workspace_uuid': old_uuid},
+            )
+            await conn.execute(
+                sa.text(
+                    "INSERT INTO metadata (key, value) VALUES ('oss_workspace_uuid', :workspace_uuid) "
+                    "ON CONFLICT(key) DO UPDATE SET value = excluded.value"
+                ),
+                {'workspace_uuid': old_uuid},
+            )
+
+        await run_alembic_upgrade(engine, 'head')
+        expected_uuid = workspace_uuid_from_instance_id(instance_uuid)
+        async with engine.connect() as conn:
+            assert await conn.scalar(sa.text("SELECT uuid FROM workspaces WHERE source = 'local'")) == expected_uuid
+            assert await conn.scalar(
+                sa.text("SELECT workspace_uuid FROM workspace_metadata WHERE key = 'migration_probe'")
+            ) == expected_uuid
+            assert await conn.scalar(
+                sa.text("SELECT value FROM metadata WHERE key = 'oss_workspace_uuid'")
+            ) == expected_uuid
+    finally:
+        await engine.dispose()
+
+
 async def test_persistence_startup_rejects_instance_uuid_drift(tmp_path, monkeypatch):
     engine = create_async_engine(f'sqlite+aiosqlite:///{tmp_path / "instance-drift.db"}')
     try:

tests/integration/persistence/test_workspace_quota_postgres.py

@@ -1,137 +0,0 @@
-"""PostgreSQL integration coverage for durable workspace quota locking.
-
-Run with TEST_POSTGRES_URL=postgresql+asyncpg://... pytest ...
-"""
-
-from __future__ import annotations
-
-import asyncio
-import os
-import uuid
-
-import pytest
-import sqlalchemy as sa
-from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
-from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
-
-from langbot.pkg.cloud import quotas as quota_module
-from langbot.pkg.cloud.quotas import WorkspaceQuota, WorkspaceQuotaExceededError, require_resource_capacity
-
-
-pytestmark = [pytest.mark.integration, pytest.mark.slow, pytest.mark.asyncio]
-
-
-class _Base(DeclarativeBase):
-    pass
-
-
-class _Workspace(_Base):
-    __tablename__ = 'quota_integration_workspaces'
-
-    uuid: Mapped[str] = mapped_column(sa.String(36), primary_key=True)
-
-
-class _Resource(_Base):
-    __tablename__ = 'quota_integration_resources'
-
-    uuid: Mapped[str] = mapped_column(sa.String(36), primary_key=True)
-    workspace_uuid: Mapped[str] = mapped_column(
-        sa.String(36),
-        sa.ForeignKey('quota_integration_workspaces.uuid', ondelete='CASCADE'),
-        nullable=False,
-        index=True,
-    )
-
-
-@pytest.fixture
-async def quota_postgres(monkeypatch):
-    url = os.environ.get('TEST_POSTGRES_URL')
-    if not url:
-        pytest.skip('TEST_POSTGRES_URL not set')
-    if url.startswith('postgresql://'):
-        url = url.replace('postgresql://', 'postgresql+asyncpg://', 1)
-
-    engine = create_async_engine(url, pool_size=5, max_overflow=0)
-    monkeypatch.setattr(quota_module.persistence_workspace, 'Workspace', _Workspace)
-    async with engine.begin() as connection:
-        await connection.run_sync(_Base.metadata.drop_all)
-        await connection.run_sync(_Base.metadata.create_all)
-    try:
-        yield url, engine
-    finally:
-        async with engine.begin() as connection:
-            await connection.run_sync(_Base.metadata.drop_all)
-        await engine.dispose()
-
-
-async def test_workspace_row_lock_is_atomic_isolated_and_survives_pool_restart(quota_postgres) -> None:
-    url, engine = quota_postgres
-    workspace_a = str(uuid.uuid4())
-    workspace_b = str(uuid.uuid4())
-    quota = WorkspaceQuota(limit=1, requires_transaction_lock=True)
-    sessions = async_sessionmaker(engine, expire_on_commit=False)
-
-    async with engine.begin() as connection:
-        await connection.execute(sa.insert(_Workspace), [{'uuid': workspace_a}, {'uuid': workspace_b}])
-
-    lock_acquired = asyncio.Event()
-    release_first = asyncio.Event()
-
-    async def admit(workspace_uuid: str, *, hold: bool = False) -> None:
-        async with sessions() as session:
-            async with session.begin():
-                await require_resource_capacity(
-                    session.execute,
-                    workspace_uuid=workspace_uuid,
-                    model=_Resource,
-                    quota=quota,
-                    resource_name='resources',
-                )
-                if hold:
-                    lock_acquired.set()
-                    await release_first.wait()
-                await session.execute(
-                    sa.insert(_Resource).values(uuid=str(uuid.uuid4()), workspace_uuid=workspace_uuid)
-                )
-
-    first = asyncio.create_task(admit(workspace_a, hold=True))
-    await asyncio.wait_for(lock_acquired.wait(), timeout=2)
-    same_workspace = asyncio.create_task(admit(workspace_a))
-    other_workspace = asyncio.create_task(admit(workspace_b))
-
-    await asyncio.wait_for(other_workspace, timeout=2)
-    assert not same_workspace.done(), 'same-workspace transaction bypassed SELECT FOR UPDATE'
-
-    release_first.set()
-    await first
-    with pytest.raises(WorkspaceQuotaExceededError, match=r'Maximum number of resources \(1\) reached'):
-        await same_workspace
-
-    async with sessions() as session:
-        counts = dict(
-            (
-                await session.execute(
-                    sa.select(_Resource.workspace_uuid, sa.func.count())
-                    .group_by(_Resource.workspace_uuid)
-                    .order_by(_Resource.workspace_uuid)
-                )
-            ).all()
-        )
-    assert counts == {workspace_a: 1, workspace_b: 1}
-
-    await engine.dispose()
-    restarted_engine = create_async_engine(url, pool_size=2, max_overflow=0)
-    restarted_sessions = async_sessionmaker(restarted_engine, expire_on_commit=False)
-    try:
-        async with restarted_sessions() as session:
-            async with session.begin():
-                with pytest.raises(WorkspaceQuotaExceededError):
-                    await require_resource_capacity(
-                        session.execute,
-                        workspace_uuid=workspace_a,
-                        model=_Resource,
-                        quota=quota,
-                        resource_name='resources',
-                    )
-    finally:
-        await restarted_engine.dispose()

tests/unit_tests/api/http/test_authz.py

@@ -27,10 +27,9 @@ def test_owner_has_every_fixed_permission():
     assert ctx.workspace.permissions == frozenset(permission.value for permission in authz.Permission)
 
 
-def test_admin_cannot_transfer_owner_delete_workspace_or_link_billing():
+def test_admin_cannot_delete_workspace_or_link_billing():
     ctx = _context(authz.WorkspaceRole.ADMIN)
 
-    assert not authz.has_permission(ctx, authz.Permission.OWNER_TRANSFER)
     assert not authz.has_permission(ctx, authz.Permission.WORKSPACE_DELETE)
     assert not authz.has_permission(ctx, authz.Permission.BILLING_LINK_MANAGE)
     assert authz.has_permission(ctx, authz.Permission.MEMBER_INVITE)

tests/unit_tests/api/http/test_internal_error_responses.py

@@ -9,7 +9,6 @@ import quart
 
 from langbot.pkg.api.http.controller import group
 from langbot.pkg.api.http.controller.groups.webhooks import WebhookRouterGroup
-from langbot.pkg.cloud.quotas import WorkspaceQuotaExceededError
 from langbot.pkg.utils.bounded_executor import (
     BlockingWorkCapacityError,
     current_blocking_work_scope,
@@ -49,16 +48,6 @@ class _BlockingCapacityRouterGroup(group.RouterGroup):
             raise BlockingWorkCapacityError('Workspace blocking executor capacity reached')
 
 
-class _QuotaRouterGroup(group.RouterGroup):
-    name = 'quota-test'
-    path = '/quota-test'
-
-    async def initialize(self) -> None:
-        @self.route('', methods=['POST'], auth_type=group.AuthType.NONE)
-        async def _():
-            raise WorkspaceQuotaExceededError('bots', 2)
-
-
 class _InvalidAccountRouterGroup(group.RouterGroup):
     name = 'invalid-account-test'
     path = '/invalid-account-test'
@@ -141,20 +130,6 @@ async def test_blocking_work_capacity_maps_to_retryable_http_response():
     }
 
 
-async def test_workspace_quota_maps_to_stable_conflict_response():
-    application = SimpleNamespace(logger=Mock())
-    quart_app = quart.Quart(__name__)
-    await _QuotaRouterGroup(application, quart_app).initialize()
-
-    response = await quart_app.test_client().post('/quota-test')
-
-    assert response.status_code == 409
-    assert await response.get_json() == {
-        'code': 'workspace_quota_exceeded',
-        'msg': 'Maximum number of bots (2) reached',
-    }
-
-
 async def test_public_webhook_carries_scope_without_holding_database_session():
     class ScopeOnlyPersistenceManager:
         mode = SimpleNamespace(value='cloud_runtime')

tests/unit_tests/api/service/test_bot_service.py

@@ -311,9 +311,10 @@ class TestBotServiceCreateBot:
         ap.platform_mgr = SimpleNamespace()
         ap.platform_mgr.load_bot = AsyncMock()
 
-        # Mock the atomic count query to report 2 existing bots.
-        mock_result = _create_mock_result()
-        mock_result.scalar_one = Mock(return_value=2)
+        # Mock get_bots to return 2 bots already
+        bot1 = _create_mock_bot(bot_uuid='uuid-1')
+        bot2 = _create_mock_bot(bot_uuid='uuid-2')
+        mock_result = _create_mock_result([bot1, bot2])
         ap.persistence_mgr.execute_async = AsyncMock(return_value=mock_result)
         ap.persistence_mgr.serialize_model = Mock(return_value={'uuid': 'uuid-1', 'name': 'Bot 1'})
 

tests/unit_tests/api/service/test_cloud_bot_quota_concurrency.py

@@ -1,129 +0,0 @@
-from __future__ import annotations
-
-import asyncio
-from collections import defaultdict
-from types import SimpleNamespace
-from unittest.mock import AsyncMock
-
-import pytest
-import sqlalchemy
-
-from langbot.pkg.api.http.service.bot import BotService
-from langbot.pkg.cloud.entitlements import EntitlementResolver, EntitlementSnapshot
-
-
-INSTANCE_UUID = 'cloud-instance'
-WORKSPACE_A = '11111111-1111-1111-1111-111111111111'
-WORKSPACE_B = '22222222-2222-2222-2222-222222222222'
-
-
-class _Provider:
-    async def get_workspace_entitlement(self, workspace_uuid: str) -> EntitlementSnapshot:
-        return EntitlementSnapshot(
-            instance_uuid=INSTANCE_UUID,
-            workspace_uuid=workspace_uuid,
-            entitlement_revision=1,
-            status='active',
-            not_before=0,
-            expires_at=4_102_444_800,
-            features={},
-            limits={'bots.max': 2},
-        )
-
-
-class _Result:
-    def __init__(self, *, first=None, scalar=None) -> None:
-        self._first = first
-        self._scalar = scalar
-
-    def first(self):
-        return self._first
-
-    def scalar_one(self):
-        return self._scalar
-
-
-class _TenantUow:
-    def __init__(self, manager: '_Persistence', workspace_uuid: str) -> None:
-        self.manager = manager
-        self.workspace_uuid = workspace_uuid
-        self.lock = manager.locks[workspace_uuid]
-
-    async def __aenter__(self):
-        await self.lock.acquire()
-        return self
-
-    async def __aexit__(self, exc_type, exc, tb):
-        self.lock.release()
-
-    async def execute(self, statement):
-        sql = str(statement)
-        if isinstance(statement, sqlalchemy.sql.dml.Insert):
-            assert statement.table.name == 'bots'
-            self.manager.bots[self.workspace_uuid].append(statement.compile().params)
-            return _Result()
-        if 'FROM workspaces' in sql:
-            assert statement._for_update_arg is not None
-            self.manager.workspace_locks_seen += 1
-            return _Result(first=(self.workspace_uuid,))
-        if 'count(' in sql.lower() and 'FROM bots' in sql:
-            return _Result(scalar=len(self.manager.bots[self.workspace_uuid]))
-        if 'FROM legacy_pipelines' in sql:
-            return _Result(first=None)
-        raise AssertionError(f'unexpected statement: {sql}')
-
-
-class _Persistence:
-    def __init__(self) -> None:
-        self.locks = defaultdict(asyncio.Lock)
-        self.bots = defaultdict(list)
-        self.workspace_locks_seen = 0
-
-    def tenant_uow(self, workspace_uuid: str) -> _TenantUow:
-        return _TenantUow(self, workspace_uuid)
-
-    async def execute_async(self, statement):
-        assert 'FROM legacy_pipelines' in str(statement)
-        return _Result(first=None)
-
-
-async def _service(manager: _Persistence) -> BotService:
-    resolver = EntitlementResolver(INSTANCE_UUID, _Provider())
-    await resolver.reconcile_active_workspaces({WORKSPACE_A, WORKSPACE_B})
-    ap = SimpleNamespace(
-        entitlement_resolver=resolver,
-        persistence_mgr=manager,
-        instance_config=SimpleNamespace(data={'system': {'limitation': {'max_bots': 99}}}),
-        platform_mgr=SimpleNamespace(load_bot=AsyncMock()),
-    )
-    service = BotService(ap)
-    service.get_bot = AsyncMock(return_value={'uuid': 'created'})
-    return service
-
-
-@pytest.mark.asyncio
-async def test_cloud_bot_quota_is_atomic_isolated_and_persists_across_service_restart() -> None:
-    manager = _Persistence()
-    service = await _service(manager)
-
-    async def create(workspace_uuid: str, index: int):
-        return await service.create_bot(workspace_uuid, {'name': f'bot-{index}'})
-
-    results = await asyncio.gather(
-        *(create(WORKSPACE_A, index) for index in range(8)),
-        *(create(WORKSPACE_B, index) for index in range(8)),
-        return_exceptions=True,
-    )
-
-    successes = [result for result in results if isinstance(result, str)]
-    failures = [result for result in results if isinstance(result, ValueError)]
-    assert len(successes) == 4
-    assert len(failures) == 12
-    assert len(manager.bots[WORKSPACE_A]) == 2
-    assert len(manager.bots[WORKSPACE_B]) == 2
-    assert manager.workspace_locks_seen == 16
-
-    restarted_service = await _service(manager)
-    with pytest.raises(ValueError, match=r'Maximum number of bots \(2\) reached'):
-        await restarted_service.create_bot(WORKSPACE_A, {'name': 'after-restart'})
-    assert len(manager.bots[WORKSPACE_A]) == 2

tests/unit_tests/api/service/test_cloud_managed_model_protection.py

@@ -0,0 +1,112 @@
+"""Cloud Runtime write protection for the managed LangBot Models catalog."""
+
+from types import SimpleNamespace
+from unittest.mock import AsyncMock
+
+import pytest
+
+from langbot.pkg.api.http.service import model as model_service_module
+from langbot.pkg.api.http.service.model import (
+    EmbeddingModelsService,
+    LLMModelsService,
+    RerankModelsService,
+    _assert_cloud_managed_provider_mutable,
+)
+from langbot.pkg.cloud.model_catalog import LANGBOT_MODELS_PROVIDER_REQUESTER
+
+
+WORKSPACE = 'workspace-a'
+PROVIDER = 'managed-provider'
+MODEL = 'managed-model'
+
+
+@pytest.mark.asyncio
+async def test_managed_provider_guard_is_cloud_only(monkeypatch) -> None:
+    async def managed_provider(_ap, _context, provider_uuid):
+        assert provider_uuid == PROVIDER
+        return {'uuid': PROVIDER, 'requester': LANGBOT_MODELS_PROVIDER_REQUESTER}
+
+    monkeypatch.setattr(model_service_module, '_require_workspace_provider', managed_provider)
+    application = SimpleNamespace(persistence_mgr=SimpleNamespace(mode=SimpleNamespace(value='cloud_runtime')))
+
+    with pytest.raises(ValueError, match='managed by Cloud'):
+        await _assert_cloud_managed_provider_mutable(
+            application,
+            WORKSPACE,
+            PROVIDER,
+        )
+
+    application.persistence_mgr.mode.value = 'normal'
+    await _assert_cloud_managed_provider_mutable(
+        application,
+        WORKSPACE,
+        PROVIDER,
+    )
+
+
+@pytest.mark.parametrize(
+    ('service_type', 'create_method', 'model_data'),
+    [
+        (LLMModelsService, 'create_llm_model', {'provider_uuid': PROVIDER, 'name': 'chat', 'abilities': []}),
+        (EmbeddingModelsService, 'create_embedding_model', {'provider_uuid': PROVIDER, 'name': 'embedding'}),
+        (RerankModelsService, 'create_rerank_model', {'provider_uuid': PROVIDER, 'name': 'rerank'}),
+    ],
+)
+@pytest.mark.asyncio
+async def test_all_model_types_reject_creation_under_managed_provider(
+    monkeypatch,
+    service_type,
+    create_method: str,
+    model_data: dict,
+) -> None:
+    guard = AsyncMock(side_effect=ValueError('LangBot Models is managed by Cloud and cannot be modified'))
+    monkeypatch.setattr(model_service_module, '_assert_cloud_managed_provider_mutable', guard)
+    application = SimpleNamespace(
+        persistence_mgr=SimpleNamespace(),
+        provider_service=SimpleNamespace(
+            get_provider=AsyncMock(return_value={'uuid': PROVIDER, 'requester': LANGBOT_MODELS_PROVIDER_REQUESTER})
+        ),
+        model_mgr=None,
+    )
+    service = service_type(application)
+
+    with pytest.raises(ValueError, match='managed by Cloud'):
+        await getattr(service, create_method)(WORKSPACE, model_data)
+
+    guard.assert_awaited_once()
+
+
+@pytest.mark.parametrize(
+    ('service_type', 'get_method', 'write_method', 'payload'),
+    [
+        (LLMModelsService, 'get_llm_model', 'update_llm_model', {'name': 'changed'}),
+        (LLMModelsService, 'get_llm_model', 'delete_llm_model', None),
+        (EmbeddingModelsService, 'get_embedding_model', 'update_embedding_model', {'name': 'changed'}),
+        (EmbeddingModelsService, 'get_embedding_model', 'delete_embedding_model', None),
+        (RerankModelsService, 'get_rerank_model', 'update_rerank_model', {'name': 'changed'}),
+        (RerankModelsService, 'get_rerank_model', 'delete_rerank_model', None),
+    ],
+)
+@pytest.mark.asyncio
+async def test_all_model_types_reject_update_and_delete_for_managed_provider(
+    monkeypatch,
+    service_type,
+    get_method: str,
+    write_method: str,
+    payload: dict | None,
+) -> None:
+    guard = AsyncMock(side_effect=ValueError('LangBot Models is managed by Cloud and cannot be modified'))
+    monkeypatch.setattr(model_service_module, '_assert_cloud_managed_provider_mutable', guard)
+    application = SimpleNamespace(persistence_mgr=SimpleNamespace(mode=SimpleNamespace(value='cloud_runtime')))
+    service = service_type(application)
+    monkeypatch.setattr(
+        service,
+        get_method,
+        AsyncMock(return_value={'uuid': MODEL, 'provider_uuid': PROVIDER, 'extra_args': {}}),
+    )
+
+    args = (WORKSPACE, MODEL) if payload is None else (WORKSPACE, MODEL, payload)
+    with pytest.raises(ValueError, match='managed by Cloud'):
+        await getattr(service, write_method)(*args)
+
+    guard.assert_awaited_once()

tests/unit_tests/api/service/test_provider_service.py

@@ -25,6 +25,7 @@ from langbot.pkg.workspace.errors import WorkspaceNotFoundError
 pytestmark = pytest.mark.asyncio
 
 WORKSPACE_UUID = 'workspace-a'
+SYSTEM_REQUESTER = 'space-chat-completions'
 
 
 def _create_mock_provider(
@@ -1005,3 +1006,56 @@ class TestProviderSecretRoundtrip:
             )
 
         ap.persistence_mgr.execute_async.assert_not_awaited()
+
+
+class TestCloudManagedProviderProtection:
+    @staticmethod
+    def _service() -> ModelProviderService:
+        ap = SimpleNamespace(
+            persistence_mgr=SimpleNamespace(
+                mode=SimpleNamespace(value='cloud_runtime'),
+                execute_async=AsyncMock(),
+            ),
+            model_mgr=SimpleNamespace(),
+        )
+        return ModelProviderService(ap)
+
+    async def test_cloud_rejects_user_created_system_requester(self):
+        service = self._service()
+
+        with pytest.raises(ValueError, match='reserved'):
+            await service.create_provider(
+                WORKSPACE_UUID,
+                {
+                    'name': 'Fake LangBot Models',
+                    'requester': SYSTEM_REQUESTER,
+                    'base_url': 'https://example.invalid/v1',
+                    'api_keys': ['fake'],
+                },
+            )
+
+        with pytest.raises(ValueError, match='reserved'):
+            await service.find_or_create_provider(
+                WORKSPACE_UUID,
+                SYSTEM_REQUESTER,
+                'https://api.langbot.cloud/v1',
+                ['fake'],
+            )
+        service.ap.persistence_mgr.execute_async.assert_not_awaited()
+
+    async def test_cloud_rejects_update_and_delete_of_managed_provider(self):
+        service = self._service()
+        service.get_provider = AsyncMock(
+            return_value={'uuid': 'system-provider', 'requester': SYSTEM_REQUESTER}
+        )
+
+        with pytest.raises(ValueError, match='managed by Cloud'):
+            await service.update_provider(WORKSPACE_UUID, 'system-provider', {'name': 'Renamed'})
+        with pytest.raises(ValueError, match='managed by Cloud'):
+            await service.delete_provider(WORKSPACE_UUID, 'system-provider')
+        service.ap.persistence_mgr.execute_async.assert_not_awaited()
+
+    async def test_oss_does_not_reserve_space_requester(self):
+        ap = SimpleNamespace(persistence_mgr=SimpleNamespace(mode=SimpleNamespace(value='oss_compat')))
+        service = ModelProviderService(ap)
+        assert service._system_requester_is_reserved(SYSTEM_REQUESTER) is False

tests/unit_tests/api/service/test_space_service.py

@@ -25,6 +25,7 @@ import time
 
 from langbot.pkg.api.http.service.space import SpaceService
 from langbot.pkg.entity.persistence.user import User
+from langbot.pkg.utils import constants
 
 
 pytestmark = pytest.mark.asyncio
@@ -573,10 +574,20 @@ class TestSpaceServiceExchangeOAuthCode:
             mock_session_obj.post.return_value.__aexit__ = AsyncMock(return_value=None)
 
             # Execute
-            result = await service.exchange_oauth_code('auth_code')
+            result = await service.exchange_oauth_code(
+                'auth_code',
+                ['workspace-1'],
+                {'workspace-1': 1_700_000_000},
+            )
 
         # Verify
         assert result['access_token'] == 'new_access_token'
+        assert mock_session_obj.post.call_args.kwargs['json'] == {
+            'code': 'auth_code',
+            'instance_id': constants.instance_id,
+            'workspace_uuids': ['workspace-1'],
+            'workspace_created_ats': {'workspace-1': 1_700_000_000},
+        }
 
     async def test_exchange_oauth_code_api_error(self):
         """Raises ValueError on API error."""

tests/unit_tests/api/service/test_user_service.py

@@ -418,6 +418,7 @@ class TestUserServiceGenerateJwtToken:
         assert token is not None
 
 
+
 class TestUserServiceVerifyJwtToken:
     """Tests for verify_jwt_token method."""
 

tests/unit_tests/api/test_adapter_session_scoping.py

@@ -149,6 +149,36 @@ async def test_session_scope_matches_exact_tenant_placement_and_principal():
     assert sessions == {}
 
 
+async def test_support_admin_sessions_are_scoped_to_the_persisted_grant():
+    def support_context(grant_jti_hash: str) -> RequestContext:
+        return RequestContext(
+            instance_uuid='instance-test',
+            placement_generation=1,
+            request_id='request-test',
+            auth_type='support-admin',
+            principal=PrincipalContext(
+                principal_type=PrincipalType.SUPPORT_ADMIN,
+                actor_account_uuid='support-actor',
+                support_session_id=grant_jti_hash,
+            ),
+            workspace=WorkspaceContext(
+                workspace_uuid='workspace-a',
+                membership_uuid=None,
+                role='owner',
+                permissions=frozenset({'resource.manage'}),
+            ),
+        )
+
+    first_context = support_context('a' * 64)
+    second_context = support_context('b' * 64)
+    sessions: dict[str, dict] = {'session-test': {'status': 'waiting'}}
+    _bind_session_scope(sessions['session-test'], first_context)
+
+    assert _get_owned_session(sessions, 'session-test', second_context) is None
+    assert _pop_owned_session(sessions, 'session-test', second_context) is None
+    assert _get_owned_session(sessions, 'session-test', first_context) is sessions['session-test']
+
+
 async def test_session_capacity_evicts_oldest_session_in_same_workspace():
     owner_context = _request_context()
     sessions: dict[str, dict] = {}

tests/unit_tests/cloud/test_bootstrap.py

@@ -66,6 +66,10 @@ class _Provider:
     def __init__(self):
         self.manifest_provider = _Manifest()
 
+    async def fetch_model_catalog(self, instance_uuid: str):
+        del instance_uuid
+        raise AssertionError('not used by bootstrap contract tests')
+
     def bootstrap(self, *, instance_uuid: str, instance_config: dict):
         del instance_config
         return VerifiedCloudDeployment(
@@ -79,6 +83,7 @@ class _Provider:
             entitlement_provider=_Entitlements(),
             directory_provider=_Directory(),
             manifest_provider=self.manifest_provider,
+            model_catalog_provider=self,
             verification_key_id='root-2026',
         )
 
@@ -103,7 +108,7 @@ def _cloud_config() -> dict:
             'use': 'pgvector',
             'pgvector': {
                 'use_business_database': True,
-                'allowed_dimensions': [384, 768, 1536],
+                'allowed_dimensions': [384, 768, 1536, 3072],
             },
         },
         'mcp': {'stdio': {'enabled': False}},
@@ -211,7 +216,6 @@ async def test_cloud_directory_capacity_contract_is_fail_closed(directory_config
     [
         ({'use_business_database': False, 'allowed_dimensions': [1536]}, 'use_business_database=true'),
         ({'use_business_database': True, 'allowed_dimensions': []}, 'allowed_dimensions'),
-        ({'use_business_database': True, 'allowed_dimensions': [3072]}, 'allowed_dimensions'),
         ({'use_business_database': True, 'allowed_dimensions': [True]}, 'allowed_dimensions'),
     ],
 )

tests/unit_tests/cloud/test_directory_projection.py

@@ -181,6 +181,39 @@ def _delta(
     )
 
 
+async def test_directory_delta_requests_model_catalog_sync_after_commit(projection_context):
+    application, _session_factory = projection_context
+    request_sync = Mock()
+    application.cloud_model_catalog_service = SimpleNamespace(request_sync=request_sync)
+    event = DirectoryEvent(
+        cursor=2,
+        uuid='20000000-0000-4000-8000-000000000002',
+        aggregate_uuid=WORKSPACE_UUID,
+        event_type='directory.changed',
+        revision=2,
+        payload={'workspace_uuid': WORKSPACE_UUID, 'directory_revision': 2},
+        created_at=datetime.datetime(2026, 7, 24, 12, 30, tzinfo=datetime.UTC),
+    )
+    batch = DirectoryEventBatch(
+        instance_uuid=INSTANCE_UUID,
+        after_cursor=1,
+        cursor=2,
+        high_water_cursor=2,
+        events=[event],
+    )
+    service = DirectoryProjectionService(
+        application,
+        _Provider([_snapshot(1)], [batch], [_delta(workspaces=[_workspace(revision=2)])]),
+        INSTANCE_UUID,
+    )
+    await service.initialize()
+    request_sync.reset_mock()
+
+    await service.sync_once()
+
+    request_sync.assert_called_once_with()
+
+
 async def test_initial_snapshot_projects_core_owned_rows(projection_context):
     application, session_factory = projection_context
     reconcile_execution_projection = Mock()
@@ -1023,7 +1056,7 @@ async def test_snapshot_for_another_instance_is_rejected(projection_context):
         await service.initialize()
 
 
-async def test_core_owned_membership_survives_directory_updates_and_omission(projection_context):
+async def test_directory_revision_zero_membership_is_adopted(projection_context):
     application, session_factory = projection_context
     service = DirectoryProjectionService(application, _Provider([_snapshot(1)]), INSTANCE_UUID)
     await service.initialize()
@@ -1034,29 +1067,125 @@ async def test_core_owned_membership_survives_directory_updates_and_omission(pro
             membership.role = 'viewer'
             membership.status = 'active'
             membership.projection_revision = 0
-            session.add(
-                WorkspaceMembership(
-                    uuid=SECOND_MEMBERSHIP_UUID,
-                    workspace_uuid=WORKSPACE_UUID,
-                    account_uuid='20000000-0000-0000-0000-000000000099',
-                    role='viewer',
-                    status='active',
-                    joined_at=membership.joined_at,
-                    projection_revision=0,
-                )
-            )
 
     projected_member = _member(revision=2).model_copy(update={'role': 'owner', 'membership_status': 'removed'})
     projected_workspace = _workspace(revision=2).model_copy(update={'members': (projected_member,)})
     await service.apply_snapshot(_snapshot(2, workspaces=[projected_workspace]))
 
     async with session_factory() as session:
-        memberships = {
-            membership.uuid: membership
-            for membership in (await session.scalars(sqlalchemy.select(WorkspaceMembership))).all()
-        }
-    assert memberships[MEMBERSHIP_UUID].role == 'viewer'
-    assert memberships[MEMBERSHIP_UUID].status == 'active'
-    assert memberships[MEMBERSHIP_UUID].projection_revision == 0
-    assert memberships[SECOND_MEMBERSHIP_UUID].status == 'active'
-    assert memberships[SECOND_MEMBERSHIP_UUID].projection_revision == 0
+        membership = await session.scalar(sqlalchemy.select(WorkspaceMembership))
+    assert membership.source == 'cloud_projection'
+    assert membership.role == 'owner'
+    assert membership.status == 'removed'
+    assert membership.projection_revision == 2
+
+
+async def test_directory_revision_zero_membership_omitted_from_snapshot_is_removed(projection_context):
+    application, session_factory = projection_context
+    service = DirectoryProjectionService(application, _Provider([_snapshot(1)]), INSTANCE_UUID)
+    await service.initialize()
+
+    historical_account_uuid = '20000000-0000-0000-0000-000000000099'
+    async with session_factory() as session:
+        async with session.begin():
+            membership = await session.scalar(sqlalchemy.select(WorkspaceMembership))
+            session.add(
+                User(
+                    uuid=historical_account_uuid,
+                    user='Historical Space Member',
+                    normalized_email='historical@example.com',
+                    password='',
+                    status='active',
+                    source='cloud_projection',
+                    projection_revision=1,
+                    account_type='space',
+                    space_account_uuid=historical_account_uuid,
+                )
+            )
+            session.add(
+                WorkspaceMembership(
+                    uuid=SECOND_MEMBERSHIP_UUID,
+                    workspace_uuid=WORKSPACE_UUID,
+                    account_uuid=historical_account_uuid,
+                    role='viewer',
+                    status='active',
+                    source='cloud_projection',
+                    joined_at=membership.joined_at,
+                    projection_revision=0,
+                )
+            )
+
+    await service.apply_snapshot(_snapshot(2))
+
+    async with session_factory() as session:
+        historical = await session.get(WorkspaceMembership, SECOND_MEMBERSHIP_UUID)
+    assert historical.status == 'removed'
+    assert historical.projection_revision == 2
+
+
+async def test_cloud_account_core_invitation_membership_survives_directory_omission(projection_context):
+    application, session_factory = projection_context
+    service = DirectoryProjectionService(application, _Provider([_snapshot(1)]), INSTANCE_UUID)
+    await service.initialize()
+
+    invited_account_uuid = '20000000-0000-0000-0000-000000000098'
+    async with session_factory() as session:
+        async with session.begin():
+            projected_membership = await session.scalar(sqlalchemy.select(WorkspaceMembership))
+            session.add(
+                User(
+                    uuid=invited_account_uuid,
+                    user='Invited Cloud Account',
+                    normalized_email='invited-cloud@example.com',
+                    password='',
+                    status='active',
+                    source='cloud_projection',
+                    projection_revision=1,
+                    account_type='space',
+                    space_account_uuid=invited_account_uuid,
+                )
+            )
+            session.add(
+                WorkspaceMembership(
+                    uuid=SECOND_MEMBERSHIP_UUID,
+                    workspace_uuid=WORKSPACE_UUID,
+                    account_uuid=invited_account_uuid,
+                    role='viewer',
+                    status='active',
+                    source='local',
+                    joined_at=projected_membership.joined_at,
+                    projection_revision=0,
+                )
+            )
+
+    await service.apply_snapshot(_snapshot(2))
+
+    async with session_factory() as session:
+        membership = await session.get(WorkspaceMembership, SECOND_MEMBERSHIP_UUID)
+    assert membership.source == 'local'
+    assert membership.status == 'active'
+    assert membership.projection_revision == 0
+
+
+async def test_directory_does_not_adopt_local_membership_with_different_uuid_for_same_cloud_account(projection_context):
+    application, session_factory = projection_context
+    service = DirectoryProjectionService(application, _Provider([_snapshot(1)]), INSTANCE_UUID)
+    await service.initialize()
+
+    async with session_factory() as session:
+        async with session.begin():
+            membership = await session.scalar(sqlalchemy.select(WorkspaceMembership))
+            membership.uuid = SECOND_MEMBERSHIP_UUID
+            membership.source = 'local'
+            membership.projection_revision = 0
+
+    projected_member = _member(revision=2).model_copy(update={'role': 'owner', 'membership_status': 'removed'})
+    projected_workspace = _workspace(revision=2).model_copy(update={'members': (projected_member,)})
+    await service.apply_snapshot(_snapshot(2, workspaces=[projected_workspace]))
+
+    async with session_factory() as session:
+        membership = await session.get(WorkspaceMembership, SECOND_MEMBERSHIP_UUID)
+    assert membership.source == 'local'
+    assert membership.role == 'developer'
+    assert membership.status == 'active'
+    assert membership.projection_revision == 0

tests/unit_tests/cloud/test_model_catalog.py

@@ -0,0 +1,466 @@
+from __future__ import annotations
+
+import asyncio
+import logging
+from datetime import UTC, datetime
+from types import SimpleNamespace
+
+import pytest
+import sqlalchemy
+from sqlalchemy.ext.asyncio import create_async_engine
+
+from langbot.pkg.cloud.model_catalog import (
+    CloudModelCatalogSnapshot,
+    CloudModelCatalogSyncService,
+    system_model_uuid,
+    system_provider_uuid,
+)
+from langbot.pkg.entity.persistence.base import Base
+from langbot.pkg.entity.persistence.model import EmbeddingModel, LLMModel, ModelProvider
+from langbot.pkg.entity.persistence.workspace import Workspace
+from langbot.pkg.persistence.mgr import PersistenceManager, PersistenceMode
+
+
+pytestmark = pytest.mark.asyncio
+INSTANCE_UUID = 'instance-model-catalog'
+WORKSPACE_A = '00000000-0000-4000-8000-000000000001'
+WORKSPACE_B = '00000000-0000-4000-8000-000000000002'
+OWNER_A = '10000000-0000-4000-8000-000000000001'
+OWNER_B = '10000000-0000-4000-8000-000000000002'
+
+
+class _CatalogProvider:
+    def __init__(self, snapshot: CloudModelCatalogSnapshot) -> None:
+        self.snapshot = snapshot
+
+    async def fetch_model_catalog(self, instance_uuid: str) -> CloudModelCatalogSnapshot:
+        assert instance_uuid == INSTANCE_UUID
+        return self.snapshot
+
+
+def _snapshot(
+    *,
+    key_a: str | None = 'owner-a-key',
+    model_id: str = 'gpt-test',
+    include_embedding: bool = True,
+) -> CloudModelCatalogSnapshot:
+    models = [
+        {
+            'uuid': 'upstream-chat',
+            'model_id': model_id,
+            'category': 'chat',
+            'llm_abilities': ['chat', 'vision'],
+            'is_featured': True,
+            'featured_order': 7,
+        }
+    ]
+    if include_embedding:
+        models.append(
+            {
+                'uuid': 'upstream-embedding',
+                'model_id': 'embedding-test',
+                'category': 'embedding',
+            }
+        )
+    return CloudModelCatalogSnapshot.model_validate(
+        {
+            'instance_uuid': INSTANCE_UUID,
+            'generated_at': datetime.now(UTC),
+            'base_url': 'https://api.langbot.cloud/v1/',
+            'models': models,
+            'workspaces': [
+                {
+                    'workspace_uuid': WORKSPACE_A,
+                    'owner_account_uuid': OWNER_A,
+                    'api_key': key_a,
+                    'credits': 25000,
+                },
+                {
+                    'workspace_uuid': WORKSPACE_B,
+                    'owner_account_uuid': OWNER_B,
+                    'api_key': 'owner-b-key',
+                    'credits': 5000,
+                },
+            ],
+        }
+    )
+
+
+async def test_catalog_snapshot_treats_null_model_abilities_as_empty() -> None:
+    payload = _snapshot().model_dump(mode='json')
+    payload['models'][0]['llm_abilities'] = None
+
+    snapshot = CloudModelCatalogSnapshot.model_validate(payload)
+
+    assert snapshot.models[0].llm_abilities == ()
+
+
+async def test_catalog_reconciles_every_workspace_idempotently_and_tracks_owner_and_downlisting(tmp_path) -> None:
+    engine = create_async_engine(f'sqlite+aiosqlite:///{tmp_path / "model-catalog.db"}')
+    manager = PersistenceManager(object(), mode=PersistenceMode.CLOUD_RUNTIME)
+    manager.db = SimpleNamespace(get_engine=lambda: engine)
+    bindings = [
+        SimpleNamespace(instance_uuid=INSTANCE_UUID, workspace_uuid=WORKSPACE_A, placement_generation=1),
+        SimpleNamespace(instance_uuid=INSTANCE_UUID, workspace_uuid=WORKSPACE_B, placement_generation=1),
+    ]
+    workspace_service = SimpleNamespace(list_active_execution_bindings=lambda: _async_value(bindings))
+    reload_counter = _AsyncCounter()
+    runtime_reload = SimpleNamespace(load_models_from_db=reload_counter)
+    app = SimpleNamespace(
+        persistence_mgr=manager,
+        workspace_service=workspace_service,
+        model_mgr=runtime_reload,
+        logger=logging.getLogger(__name__),
+    )
+    provider = _CatalogProvider(_snapshot())
+    service = CloudModelCatalogSyncService(app, provider, INSTANCE_UUID)
+
+    try:
+        async with engine.begin() as connection:
+            await connection.run_sync(Base.metadata.create_all)
+            await connection.execute(
+                sqlalchemy.insert(Workspace),
+                [
+                    {
+                        'uuid': WORKSPACE_A,
+                        'instance_uuid': INSTANCE_UUID,
+                        'name': 'A',
+                        'slug': 'a',
+                        'source': 'cloud_projection',
+                    },
+                    {
+                        'uuid': WORKSPACE_B,
+                        'instance_uuid': INSTANCE_UUID,
+                        'name': 'B',
+                        'slug': 'b',
+                        'source': 'cloud_projection',
+                    },
+                ],
+            )
+            await connection.execute(
+                sqlalchemy.insert(ModelProvider).values(
+                    uuid='custom-provider',
+                    workspace_uuid=WORKSPACE_A,
+                    name='Custom',
+                    requester='openai-chat-completions',
+                    base_url='https://custom.example/v1',
+                    api_keys=['custom-key'],
+                )
+            )
+            await connection.execute(
+                sqlalchemy.insert(LLMModel).values(
+                    uuid='custom-model',
+                    workspace_uuid=WORKSPACE_A,
+                    name='custom-model',
+                    provider_uuid='custom-provider',
+                    abilities=['chat'],
+                    extra_args={},
+                    prefered_ranking=0,
+                )
+            )
+
+        first = await service.sync_once()
+        assert first == {'workspaces': 2, 'created': 6, 'updated': 0, 'deleted': 0}
+        assert reload_counter.calls == 1
+        assert service.get_workspace_credits(WORKSPACE_A) == 25000
+        assert service.get_workspace_credits(WORKSPACE_B) == 5000
+
+        async with engine.connect() as connection:
+            providers = (
+                await connection.execute(
+                    sqlalchemy.select(
+                        ModelProvider.uuid,
+                        ModelProvider.workspace_uuid,
+                        ModelProvider.api_keys,
+                    ).where(ModelProvider.requester == 'space-chat-completions')
+                )
+            ).all()
+            assert {item.workspace_uuid for item in providers} == {WORKSPACE_A, WORKSPACE_B}
+            assert {item.uuid for item in providers} == {
+                system_provider_uuid(WORKSPACE_A),
+                system_provider_uuid(WORKSPACE_B),
+            }
+            assert {item.workspace_uuid: item.api_keys for item in providers} == {
+                WORKSPACE_A: ['owner-a-key'],
+                WORKSPACE_B: ['owner-b-key'],
+            }
+            assert await connection.scalar(sqlalchemy.select(sqlalchemy.func.count()).select_from(LLMModel)) == 3
+            assert await connection.scalar(sqlalchemy.select(sqlalchemy.func.count()).select_from(EmbeddingModel)) == 2
+
+        second = await service.sync_once()
+        assert second == {'workspaces': 2, 'created': 0, 'updated': 0, 'deleted': 0}
+        assert reload_counter.calls == 1
+
+        provider.snapshot = _snapshot(
+            key_a='new-owner-key',
+            model_id='gpt-renamed',
+            include_embedding=False,
+        )
+        third = await service.sync_once()
+        assert third == {'workspaces': 2, 'created': 0, 'updated': 3, 'deleted': 2}
+        assert reload_counter.calls == 2
+
+        async with engine.connect() as connection:
+            provider_a_keys = await connection.scalar(
+                sqlalchemy.select(ModelProvider.api_keys).where(ModelProvider.uuid == system_provider_uuid(WORKSPACE_A))
+            )
+            assert provider_a_keys == ['new-owner-key']
+            system_model_names = (
+                (
+                    await connection.execute(
+                        sqlalchemy.select(LLMModel.name).where(
+                            LLMModel.provider_uuid.in_(
+                                [system_provider_uuid(WORKSPACE_A), system_provider_uuid(WORKSPACE_B)]
+                            )
+                        )
+                    )
+                )
+                .scalars()
+                .all()
+            )
+            assert set(system_model_names) == {'gpt-renamed'}
+            assert await connection.scalar(sqlalchemy.select(sqlalchemy.func.count()).select_from(EmbeddingModel)) == 0
+            assert (
+                await connection.scalar(
+                    sqlalchemy.select(sqlalchemy.func.count())
+                    .select_from(ModelProvider)
+                    .where(ModelProvider.uuid == 'custom-provider')
+                )
+                == 1
+            )
+            assert (
+                await connection.scalar(
+                    sqlalchemy.select(sqlalchemy.func.count())
+                    .select_from(LLMModel)
+                    .where(LLMModel.uuid == 'custom-model')
+                )
+                == 1
+            )
+
+        provider.snapshot = _snapshot(key_a=None, model_id='gpt-renamed', include_embedding=False)
+        fourth = await service.sync_once()
+        assert fourth == {'workspaces': 2, 'created': 0, 'updated': 1, 'deleted': 0}
+        assert reload_counter.calls == 3
+        async with engine.connect() as connection:
+            provider_a_keys = await connection.scalar(
+                sqlalchemy.select(ModelProvider.api_keys).where(ModelProvider.uuid == system_provider_uuid(WORKSPACE_A))
+            )
+            assert provider_a_keys == []
+    finally:
+        await engine.dispose()
+
+
+def test_workspace_scoped_ids_are_stable_and_secrets_are_redacted() -> None:
+    assert system_provider_uuid(WORKSPACE_A) == system_provider_uuid(WORKSPACE_A)
+    assert system_provider_uuid(WORKSPACE_A) != system_provider_uuid(WORKSPACE_B)
+    assert system_model_uuid(WORKSPACE_A, 'chat', 'upstream') != system_model_uuid(WORKSPACE_B, 'chat', 'upstream')
+    snapshot = _snapshot()
+    assert 'owner-a-key' not in repr(snapshot)
+
+
+async def test_snapshot_must_cover_every_active_workspace() -> None:
+    snapshot = _snapshot().model_copy(update={'workspaces': _snapshot().workspaces[:1]})
+    app = SimpleNamespace(
+        workspace_service=SimpleNamespace(
+            list_active_execution_bindings=lambda: _async_value(
+                [SimpleNamespace(workspace_uuid=WORKSPACE_A), SimpleNamespace(workspace_uuid=WORKSPACE_B)]
+            )
+        ),
+        logger=logging.getLogger(__name__),
+    )
+    service = CloudModelCatalogSyncService(app, _CatalogProvider(snapshot), INSTANCE_UUID)
+    with pytest.raises(ValueError, match='missing billing projections for 1 active Workspaces'):
+        await service.sync_once()
+
+
+async def test_periodic_sync_discovers_workspace_created_after_startup_cache_release(tmp_path) -> None:
+    engine = create_async_engine(f'sqlite+aiosqlite:///{tmp_path / "model-catalog-new-workspace.db"}')
+    manager = PersistenceManager(object(), mode=PersistenceMode.CLOUD_RUNTIME)
+    manager.db = SimpleNamespace(get_engine=lambda: engine)
+    startup_bindings = [
+        SimpleNamespace(instance_uuid=INSTANCE_UUID, workspace_uuid=WORKSPACE_A, placement_generation=1)
+    ]
+    live_bindings = [
+        *startup_bindings,
+        SimpleNamespace(instance_uuid=INSTANCE_UUID, workspace_uuid=WORKSPACE_B, placement_generation=1),
+    ]
+
+    class _WorkspaceService:
+        startup_released = False
+
+        async def list_active_execution_bindings(self):
+            return list(live_bindings if self.startup_released else startup_bindings)
+
+        def release_startup_execution_bindings(self):
+            self.startup_released = True
+
+    workspace_service = _WorkspaceService()
+    app = SimpleNamespace(
+        persistence_mgr=manager,
+        workspace_service=workspace_service,
+        model_mgr=SimpleNamespace(load_models_from_db=_AsyncCounter()),
+        logger=logging.getLogger(__name__),
+    )
+    service = CloudModelCatalogSyncService(app, _CatalogProvider(_snapshot()), INSTANCE_UUID)
+
+    try:
+        async with engine.begin() as connection:
+            await connection.run_sync(Base.metadata.create_all)
+            await connection.execute(
+                sqlalchemy.insert(Workspace),
+                [
+                    {
+                        'uuid': WORKSPACE_A,
+                        'instance_uuid': INSTANCE_UUID,
+                        'name': 'A',
+                        'slug': 'a',
+                        'source': 'cloud_projection',
+                    },
+                    {
+                        'uuid': WORKSPACE_B,
+                        'instance_uuid': INSTANCE_UUID,
+                        'name': 'B',
+                        'slug': 'b',
+                        'source': 'cloud_projection',
+                    },
+                ],
+            )
+
+        await service.initialize()
+        workspace_service.release_startup_execution_bindings()
+        await service.sync_once()
+
+        async with engine.connect() as connection:
+            provider_b = await connection.scalar(
+                sqlalchemy.select(ModelProvider).where(ModelProvider.uuid == system_provider_uuid(WORKSPACE_B))
+            )
+        assert provider_b is not None
+    finally:
+        await engine.dispose()
+
+
+async def test_catalog_run_wakes_immediately_when_directory_changes() -> None:
+    sync_started = asyncio.Event()
+
+    class _WakeService(CloudModelCatalogSyncService):
+        async def sync_once(self, *, reload_runtime: bool = True):
+            del reload_runtime
+            sync_started.set()
+            return {'workspaces': 0, 'created': 0, 'updated': 0, 'deleted': 0}
+
+    app = SimpleNamespace(logger=logging.getLogger(__name__))
+    service = _WakeService(app, _CatalogProvider(_snapshot()), INSTANCE_UUID, sync_interval_seconds=3600)
+    task = asyncio.create_task(service.run())
+    try:
+        await asyncio.sleep(0)
+        service.request_sync()
+        await asyncio.wait_for(sync_started.wait(), timeout=0.2)
+    finally:
+        task.cancel()
+        with pytest.raises(asyncio.CancelledError):
+            await task
+
+
+async def _async_value(value):
+    return value
+
+
+class _AsyncCounter:
+    def __init__(self) -> None:
+        self.calls = 0
+
+    async def __call__(self) -> None:
+        self.calls += 1
+
+
+async def test_partial_workspace_failure_reloads_already_committed_changes() -> None:
+    bindings = [
+        SimpleNamespace(workspace_uuid=WORKSPACE_A),
+        SimpleNamespace(workspace_uuid=WORKSPACE_B),
+    ]
+    reload_counter = _AsyncCounter()
+    app = SimpleNamespace(
+        workspace_service=SimpleNamespace(list_active_execution_bindings=lambda: _async_value(bindings)),
+        model_mgr=SimpleNamespace(load_models_from_db=reload_counter),
+        logger=logging.getLogger(__name__),
+    )
+    service = CloudModelCatalogSyncService(app, _CatalogProvider(_snapshot()), INSTANCE_UUID)
+    calls = 0
+
+    async def sync_workspace(*_args):
+        nonlocal calls
+        calls += 1
+        if calls == 1:
+            return {'created': 1, 'updated': 0, 'deleted': 0}
+        raise RuntimeError('second Workspace failed')
+
+    service._sync_workspace = sync_workspace  # type: ignore[method-assign]
+
+    with pytest.raises(RuntimeError, match='second Workspace failed'):
+        await service.sync_once()
+    assert service.get_workspace_credits(WORKSPACE_A) == 25000
+    assert service.get_workspace_credits(WORKSPACE_B) is None
+    assert reload_counter.calls == 1
+
+
+async def test_failed_runtime_reload_is_retried_after_noop_sync() -> None:
+    bindings = [SimpleNamespace(workspace_uuid=WORKSPACE_A)]
+
+    class _FlakyReload:
+        def __init__(self) -> None:
+            self.calls = 0
+
+        async def __call__(self) -> None:
+            self.calls += 1
+            if self.calls == 1:
+                raise RuntimeError('reload failed')
+
+    runtime_reload = _FlakyReload()
+    app = SimpleNamespace(
+        workspace_service=SimpleNamespace(list_active_execution_bindings=lambda: _async_value(bindings)),
+        model_mgr=SimpleNamespace(load_models_from_db=runtime_reload),
+        logger=logging.getLogger(__name__),
+    )
+    service = CloudModelCatalogSyncService(app, _CatalogProvider(_snapshot()), INSTANCE_UUID)
+    calls = 0
+
+    async def sync_workspace(*_args):
+        nonlocal calls
+        calls += 1
+        if calls == 1:
+            return {'created': 1, 'updated': 0, 'deleted': 0}
+        return {'created': 0, 'updated': 0, 'deleted': 0}
+
+    service._sync_workspace = sync_workspace  # type: ignore[method-assign]
+
+    with pytest.raises(RuntimeError, match='reload failed'):
+        await service.sync_once()
+    summary = await service.sync_once()
+    assert summary == {'workspaces': 1, 'created': 0, 'updated': 0, 'deleted': 0}
+    assert runtime_reload.calls == 2
+
+
+async def test_background_sync_log_redacts_exception_message(caplog) -> None:
+    secret = 'owner-secret-api-key'
+    attempted = asyncio.Event()
+
+    class _FailingProvider:
+        async def fetch_model_catalog(self, instance_uuid: str) -> CloudModelCatalogSnapshot:
+            del instance_uuid
+            attempted.set()
+            raise RuntimeError(f'database parameters include {secret}')
+
+    app = SimpleNamespace(logger=logging.getLogger(__name__))
+    service = CloudModelCatalogSyncService(app, _FailingProvider(), INSTANCE_UUID)
+    service.sync_interval_seconds = 0.001
+    task = asyncio.create_task(service.run())
+    try:
+        await asyncio.wait_for(attempted.wait(), timeout=1)
+        await asyncio.sleep(0.01)
+    finally:
+        task.cancel()
+        with pytest.raises(asyncio.CancelledError):
+            await task
+
+    assert secret not in caplog.text
+    assert 'Cloud model catalog synchronization failed (RuntimeError)' in caplog.text

tests/unit_tests/cloud/test_space_launch.py

@@ -11,6 +11,7 @@ from cryptography.hazmat.primitives import serialization
 from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
 
 from langbot.pkg.cloud.launch import SpaceLaunchError, SpaceLaunchService
+from langbot.pkg.cloud.support_admin import SupportAdminReplayError
 
 
 pytestmark = pytest.mark.asyncio
@@ -48,7 +49,6 @@ def _claims(*, now: int, jti: str | None = None, workspace_uuid: str = WORKSPACE
         'payload': {
             'account_uuid': ACCOUNT_UUID,
             'workspace_uuid': workspace_uuid,
-            'return_path': '/',
         },
     }
 
@@ -58,9 +58,21 @@ def _service(private_key: Ed25519PrivateKey, *, now: int) -> SpaceLaunchService:
         encoding=serialization.Encoding.Raw,
         format=serialization.PublicFormat.Raw,
     )
+    consumed: set[str] = set()
+
+    class DurableSupportAdminService:
+        async def consume_launch_grant(self, **kwargs):
+            grant_hash = kwargs['grant_jti_hash']
+            if grant_hash in consumed:
+                raise SupportAdminReplayError('already consumed')
+            consumed.add(grant_hash)
+            return SimpleNamespace(token='support-admin-token')
+
     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),
+        support_admin_session_service=DurableSupportAdminService(),
         instance_config=SimpleNamespace(
             data={
                 'space': {
@@ -82,30 +94,85 @@ async def test_consumes_valid_workspace_launch_assertion_once():
 
     launch = await service.consume_assertion(token, expected_workspace_uuid=WORKSPACE_UUID)
 
-    assert launch == {
-        'account_uuid': ACCOUNT_UUID,
-        'workspace_uuid': WORKSPACE_UUID,
-        'return_path': '/',
-    }
+    assert launch == {'account_uuid': ACCOUNT_UUID, 'workspace_uuid': WORKSPACE_UUID}
     with pytest.raises(SpaceLaunchError, match='already been consumed'):
         await service.consume_assertion(token, expected_workspace_uuid=WORKSPACE_UUID)
 
 
-async def test_consumed_assertion_remains_blocked_through_clock_skew_window():
+
+
+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['iat'] = now - 10
-    claims['nbf'] = now - 10
-    claims['exp'] = now - 1
+    claims['kind'] = 'workspace.support_admin_launch'
+    claims['payload'].update(
+        {
+            'launch_mode': 'support_admin',
+            'principal_type': 'support_admin',
+            'actor_account_uuid': '33333333-3333-4333-8333-333333333333',
+            'effective_role': 'owner',
+        }
+    )
+    claims['payload'].pop('account_uuid')
     token = _sign(private_key, claims)
 
-    await service.consume_assertion(token, expected_workspace_uuid=WORKSPACE_UUID)
+    launch = await service.consume_assertion(token, expected_workspace_uuid=WORKSPACE_UUID)
 
+    assert launch == {
+        'workspace_uuid': WORKSPACE_UUID,
+        'launch_mode': 'support_admin',
+        'actor_account_uuid': '33333333-3333-4333-8333-333333333333',
+        'effective_role': 'owner',
+        'grant_jti_hash': launch['grant_jti_hash'],
+        'support_admin_token': 'support-admin-token',
+    }
     with pytest.raises(SpaceLaunchError, match='already been consumed'):
         await service.consume_assertion(token, expected_workspace_uuid=WORKSPACE_UUID)
 
+    invalid = _claims(now=now)
+    invalid['kind'] = 'workspace.support_admin_launch'
+    invalid['payload'].update(
+        {
+            'launch_mode': 'support_admin',
+            'principal_type': 'support_admin',
+            'actor_account_uuid': '33333333-3333-4333-8333-333333333333',
+            'effective_role': 'member',
+        }
+    )
+    invalid['payload'].pop('account_uuid')
+    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['kind'] = 'workspace.support_admin_launch'
+    too_long['exp'] = now + 91
+    too_long['payload'].update(
+        {
+            'launch_mode': 'support_admin',
+            'principal_type': 'support_admin',
+            'actor_account_uuid': '33333333-3333-4333-8333-333333333333',
+            'effective_role': 'owner',
+        }
+    )
+    too_long['payload'].pop('account_uuid')
+    with pytest.raises(SpaceLaunchError, match='lifetime exceeds 90 seconds'):
+        await service.consume_assertion(_sign(private_key, too_long), expected_workspace_uuid=WORKSPACE_UUID)
+
+    impersonating = _claims(now=now)
+    impersonating['kind'] = 'workspace.support_admin_launch'
+    impersonating['payload'].update(
+        {
+            'launch_mode': 'support_admin',
+            'principal_type': 'support_admin',
+            'actor_account_uuid': '33333333-3333-4333-8333-333333333333',
+            'effective_role': 'owner',
+        }
+    )
+    with pytest.raises(SpaceLaunchError, match='customer Account'):
+        await service.consume_assertion(_sign(private_key, impersonating), expected_workspace_uuid=WORKSPACE_UUID)
+
 
 async def test_replay_cache_does_not_scan_all_live_assertions(monkeypatch):
     private_key = Ed25519PrivateKey.generate()
@@ -183,15 +250,3 @@ async def test_rejects_invalid_signature_and_non_cloud_mode():
     oss_service.ap.deployment.multi_workspace_enabled = False
     with pytest.raises(SpaceLaunchError, match='verified Cloud mode'):
         await oss_service.consume_assertion(token, expected_workspace_uuid=WORKSPACE_UUID)
-
-
-@pytest.mark.asyncio
-async def test_rejects_unsafe_signed_return_path() -> None:
-    private_key = Ed25519PrivateKey.generate()
-    now = int(time.time())
-    service = _service(private_key, now=now)
-    claims = _claims(now=now)
-    claims['payload']['return_path'] = '//evil.example'
-
-    with pytest.raises(SpaceLaunchError, match='return path'):
-        await service.consume_assertion(_sign(private_key, claims))

tests/unit_tests/cloud/test_workspace_quotas.py

@@ -1,102 +0,0 @@
-from __future__ import annotations
-
-from types import SimpleNamespace
-from unittest.mock import AsyncMock, Mock
-
-import pytest
-import sqlalchemy
-
-from langbot.pkg.cloud.entitlements import EntitlementResolver, EntitlementSnapshot
-from langbot.pkg.cloud.quotas import (
-    WorkspaceQuota,
-    require_resource_capacity,
-    resolve_workspace_quota,
-)
-from langbot.pkg.entity.persistence.bot import Bot
-
-
-WORKSPACE_UUID = '11111111-1111-1111-1111-111111111111'
-INSTANCE_UUID = 'cloud-instance'
-
-
-class _Provider:
-    def __init__(self, limits: dict[str, int]) -> None:
-        self.limits = limits
-
-    async def get_workspace_entitlement(self, workspace_uuid: str) -> EntitlementSnapshot:
-        return EntitlementSnapshot(
-            instance_uuid=INSTANCE_UUID,
-            workspace_uuid=workspace_uuid,
-            entitlement_revision=1,
-            status='active',
-            not_before=0,
-            expires_at=4_102_444_800,
-            features={},
-            limits=self.limits,
-            plan_name='test',
-        )
-
-
-async def _resolver(limits: dict[str, int]) -> EntitlementResolver:
-    resolver = EntitlementResolver(INSTANCE_UUID, _Provider(limits))
-    await resolver.reconcile_active_workspaces({WORKSPACE_UUID})
-    return resolver
-
-
-@pytest.mark.asyncio
-async def test_resolve_workspace_quota_uses_signed_cloud_limit() -> None:
-    ap = SimpleNamespace(entitlement_resolver=await _resolver({'bots.max': 2}))
-
-    quota = await resolve_workspace_quota(ap, WORKSPACE_UUID, 'bots.max', fallback=99)
-
-    assert quota == WorkspaceQuota(limit=2, requires_transaction_lock=True)
-
-
-@pytest.mark.asyncio
-async def test_resolve_workspace_quota_preserves_oss_fallback() -> None:
-    quota = await resolve_workspace_quota(SimpleNamespace(), WORKSPACE_UUID, 'bots.max', fallback=7)
-
-    assert quota == WorkspaceQuota(limit=7, requires_transaction_lock=False)
-
-
-@pytest.mark.asyncio
-async def test_require_resource_capacity_locks_workspace_before_counting() -> None:
-    statements: list[object] = []
-    lock_result = Mock()
-    lock_result.first.return_value = (WORKSPACE_UUID,)
-    count_result = Mock()
-    count_result.scalar_one.return_value = 1
-    execute = AsyncMock(side_effect=[lock_result, count_result])
-
-    await require_resource_capacity(
-        execute,
-        workspace_uuid=WORKSPACE_UUID,
-        model=Bot,
-        quota=WorkspaceQuota(limit=2, requires_transaction_lock=True),
-        resource_name='bots',
-    )
-
-    statements.extend(call.args[0] for call in execute.await_args_list)
-    assert len(statements) == 2
-    assert isinstance(statements[0], sqlalchemy.sql.Select)
-    assert statements[0]._for_update_arg is not None
-    assert 'workspaces' in str(statements[0])
-    assert 'count' in str(statements[1]).lower()
-
-
-@pytest.mark.asyncio
-async def test_require_resource_capacity_rejects_at_boundary() -> None:
-    lock_result = Mock()
-    lock_result.first.return_value = (WORKSPACE_UUID,)
-    count_result = Mock()
-    count_result.scalar_one.return_value = 2
-    execute = AsyncMock(side_effect=[lock_result, count_result])
-
-    with pytest.raises(ValueError, match=r'Maximum number of bots \(2\) reached'):
-        await require_resource_capacity(
-            execute,
-            workspace_uuid=WORKSPACE_UUID,
-            model=Bot,
-            quota=WorkspaceQuota(limit=2, requires_transaction_lock=True),
-            resource_name='bots',
-        )

tests/unit_tests/core/test_load_config.py

@@ -319,6 +319,7 @@ class TestApplyEnvOverridesToConfig:
         load_config = get_load_config_module()
         cfg = {
             'plugin': {
+                'connect_timeout_seconds': 30.0,
                 'worker': {
                     'max_cpus': 1.0,
                     'max_memory_mb': 512,
@@ -329,11 +330,12 @@ class TestApplyEnvOverridesToConfig:
                     'restart_failure_threshold': 8,
                     'restart_failure_window_seconds': 30.0,
                     'restart_circuit_open_seconds': 60.0,
-                }
+                },
             },
             'mcp': {'stdio': {'enabled': True}},
         }
         env = {
+            'PLUGIN__CONNECT_TIMEOUT_SECONDS': '180',
             'PLUGIN__WORKER__MAX_CPUS': '2.5',
             'PLUGIN__WORKER__MAX_MEMORY_MB': '1024',
             'PLUGIN__WORKER__MAX_PIDS': '64',
@@ -349,6 +351,7 @@ class TestApplyEnvOverridesToConfig:
         with patch.dict(os.environ, env, clear=True):
             result = load_config._apply_env_overrides_to_config(cfg)
 
+        assert result['plugin']['connect_timeout_seconds'] == 180.0
         assert result['plugin']['worker'] == {
             'max_cpus': 2.5,
             'max_memory_mb': 1024,
@@ -393,6 +396,14 @@ class TestApplyEnvOverridesToConfig:
         assert isinstance(result['plugin']['worker']['max_memory_mb'], int)
         assert result['mcp']['stdio']['enabled'] is False
 
+    def test_runtime_policy_defaults_add_typed_plugin_connect_timeout(self):
+        load_config = get_load_config_module()
+
+        completed = load_config._complete_runtime_policy_defaults({'plugin': {'enable': True}})
+
+        assert completed['plugin']['connect_timeout_seconds'] == 180.0
+        assert isinstance(completed['plugin']['connect_timeout_seconds'], float)
+
     def test_webhook_prefix_override(self):
         """Test overriding webhook_prefix via environment variable."""
         load_config = get_load_config_module()

tests/unit_tests/persistence/test_tenant_uow.py

@@ -7,7 +7,7 @@ from types import SimpleNamespace
 
 import pytest
 import sqlalchemy as sa
-from pgvector.sqlalchemy import Vector
+from pgvector.sqlalchemy import HALFVEC, Vector
 from sqlalchemy.dialects.postgresql import insert as postgresql_insert
 from sqlalchemy.dialects.sqlite import insert as sqlite_insert
 from sqlalchemy.ext.asyncio import create_async_engine
@@ -961,11 +961,13 @@ async def test_scoped_session_rejects_raw_or_unapproved_sql(
         sa.select(sa.func.coalesce(sa.func.sum(sa.literal(1)), sa.literal(0))),
         sa.select(
             sa.func.now(),
+            sa.func.date_trunc('hour', sa.column('timestamp')),
             sa.func.length(sa.literal('value')),
             sa.func.nullif(sa.literal('value'), sa.literal('')),
         ),
         sa.select(sa.column('embedding').op('<=>')(sa.literal([0.1]))),
         sa.select(sa.cast(sa.column('embedding'), Vector(384))),
+        sa.select(sa.cast(sa.column('embedding'), HALFVEC(3072))),
         sa.insert(sa.table('rows', sa.column('id'))).values(id=1),
         _multi_value_statement(value=1),
         _on_conflict_statement(update_value=sa.func.coalesce(sa.literal(1), sa.literal(0))),

tests/unit_tests/pipeline/test_controller_tenancy.py

@@ -11,6 +11,7 @@ from sqlalchemy.ext.asyncio import create_async_engine
 from langbot.pkg.persistence.mgr import PersistenceManager, PersistenceMode
 from langbot.pkg.persistence.tenant_uow import PersistenceScopeKind
 from langbot.pkg.pipeline.controller import Controller
+from langbot.pkg.pipeline.pool import QueryPool
 from langbot.pkg.workspace.errors import WorkspaceGenerationMismatchError
 
 
@@ -143,3 +144,31 @@ async def test_controller_revalidates_generation_before_running_pipeline(
     runtime_pipeline.run.assert_awaited_once_with(sample_query)
     query_pool.remove_query.assert_awaited_once_with(sample_query)
     session._semaphore.release.assert_called_once_with()
+
+
+@pytest.mark.asyncio
+async def test_controller_schedules_query_without_removing_it_twice(mock_app, sample_query):
+    query_pool = QueryPool()
+    query_pool.queries.append(sample_query)
+    mock_app.query_pool = query_pool
+    mock_app.sess_mgr.get_session = AsyncMock(return_value=SimpleNamespace(_semaphore=asyncio.Semaphore(1)))
+
+    scheduler_errors: list[str] = []
+
+    def stop_on_scheduler_error(message):
+        scheduler_errors.append(str(message))
+        raise asyncio.CancelledError
+
+    def stop_after_scheduling(process_coro, **_kwargs):
+        process_coro.close()
+        raise asyncio.CancelledError
+
+    mock_app.logger.error.side_effect = stop_on_scheduler_error
+    mock_app.task_mgr.create_task.side_effect = stop_after_scheduling
+    controller = Controller(mock_app)
+
+    with pytest.raises(asyncio.CancelledError):
+        await controller.consumer()
+
+    assert scheduler_errors == []
+    assert query_pool.queries == []

tests/unit_tests/platform/test_telegram_adapter.py

@@ -1,7 +1,8 @@
 """Tests for Telegram Dify form callback helpers."""
 
 import json
-from unittest.mock import AsyncMock, MagicMock
+from contextlib import asynccontextmanager
+from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
 from telegram import ForceReply
@@ -9,8 +10,10 @@ from telegram import ForceReply
 import langbot_plugin.api.entities.builtin.platform.entities as platform_entities
 import langbot_plugin.api.entities.builtin.platform.events as platform_events
 import langbot_plugin.api.entities.builtin.platform.message as platform_message
+
 from langbot.pkg.platform.sources.telegram import (
     TelegramAdapter,
+    TelegramMessageConverter,
     _decode_telegram_base64_limited,
     _telegram_form_action_from_callback,
     _telegram_select_field_options,
@@ -26,6 +29,70 @@ def test_telegram_base64_decode_is_bounded(monkeypatch):
         _decode_telegram_base64_limited('A' * 12)
 
 
+TELEGRAM_BOT_TOKEN = '123456789:AAExampleBotTokenThatMustNotLeak'
+TELEGRAM_FILE_URL = f'https://api.telegram.org/file/bot{TELEGRAM_BOT_TOKEN}/photos/file_0.jpg'
+
+
+@pytest.mark.asyncio
+async def test_telegram_photo_does_not_expose_bot_token_in_image_url():
+    """Regression test for the Telegram bot-token leak.
+
+    telegram.Bot builds file.file_path as
+    https://api.telegram.org/file/bot<TOKEN>/<path>, embedding the bot token.
+    The converter must not copy that URL into Image.url, or the token leaks to
+    the monitoring DB, dashboard and every installed plugin via the message
+    chain. Only base64 (which carries no token) may be stored.
+    """
+    tg_file = MagicMock()
+    tg_file.file_path = TELEGRAM_FILE_URL
+
+    photo_size = MagicMock()
+    photo_size.get_file = AsyncMock(return_value=tg_file)
+
+    message = MagicMock()
+    message.text = None
+    message.caption = None
+    message.photo = [photo_size]
+    message.voice = None
+    message.document = None
+
+    response = MagicMock()
+    response.headers = {}
+
+    async def iter_chunked(_chunk_size):
+        yield b'\xff\xd8\xff\xe0jpeg-bytes'
+
+    response.content.iter_chunked = iter_chunked
+
+    @asynccontextmanager
+    async def fake_get(url):
+        yield response
+
+    fake_session = MagicMock()
+    fake_session.get = fake_get
+
+    with patch(
+        'langbot.pkg.platform.sources.telegram.httpclient.get_session',
+        return_value=fake_session,
+    ):
+        chain = await TelegramMessageConverter.target2yiri(message, MagicMock(), 'bot-account')
+
+    images = [c for c in chain if isinstance(c, platform_message.Image)]
+    assert len(images) == 1
+    image = images[0]
+
+    # The token-bearing URL must not be retained anywhere on the component.
+    assert not image.url
+    assert image.base64 is not None
+    assert image.base64.startswith('data:image/jpeg;base64,')
+
+    # Belt-and-suspenders: the token must not appear in the serialized chain
+    # (this is what gets persisted to the monitoring DB and sent to plugins).
+    serialized = json.dumps(chain.model_dump(), ensure_ascii=False)
+    assert TELEGRAM_BOT_TOKEN not in serialized
+    assert 'api.telegram.org/file/bot' not in serialized
+
+
 def _select_form_data() -> dict:
     return {
         '_current_input_field': 'choice',

tests/unit_tests/platform/test_websocket_session_isolation.py

@@ -1,6 +1,7 @@
 """Regression tests for isolated embed-widget conversations."""
 
 import asyncio
+import contextvars
 from pathlib import Path
 from unittest.mock import AsyncMock, Mock
 
@@ -204,6 +205,48 @@ async def test_embed_event_uses_stable_session_launcher(monkeypatch):
     assert received[0].sender.id == f'websocket_pipeline-1:{session_id}'
 
 
+@pytest.mark.asyncio
+async def test_pipeline_override_survives_detached_listener_task(monkeypatch):
+    manager = WebSocketConnectionManager()
+    connection = await manager.add_connection(
+        websocket=Mock(),
+        scope=SCOPE_A,
+        pipeline_uuid='pipeline-1',
+        session_type='person',
+    )
+    monkeypatch.setattr(websocket_adapter_module, 'ws_connection_manager', manager)
+
+    class DetachedTaskManager:
+        def __init__(self):
+            self.tasks = []
+
+        def create_task(self, coro, **_kwargs):
+            task = asyncio.create_task(coro, context=contextvars.Context())
+            self.tasks.append(task)
+            return Mock(task=task)
+
+    task_manager = DetachedTaskManager()
+    adapter = WebSocketAdapter.model_construct(
+        ap=Mock(task_mgr=task_manager),
+        logger=_adapter_logger(),
+    )
+    adapter.websocket_person_session = WebSocketSession(id='person')
+    adapter.websocket_group_session = WebSocketSession(id='group')
+    pipeline_overrides = []
+
+    async def listener(_event, callback_adapter):
+        pipeline_overrides.append(callback_adapter.get_pipeline_uuid_override())
+
+    adapter.listeners = {platform_events.FriendMessage: listener}
+    await adapter.handle_websocket_message(
+        connection,
+        {'message': [{'type': 'Plain', 'text': 'hello'}], 'stream': False},
+    )
+    await asyncio.gather(*task_manager.tasks)
+
+    assert pipeline_overrides == ['pipeline-1']
+
+
 @pytest.mark.asyncio
 async def test_embed_group_event_uses_stable_session_launcher(monkeypatch):
     manager = WebSocketConnectionManager()

tests/unit_tests/plugin/test_cloud_plugin_quota_concurrency.py

@@ -1,191 +0,0 @@
-from __future__ import annotations
-
-import asyncio
-from collections import defaultdict
-from types import SimpleNamespace
-
-import pytest
-import sqlalchemy
-
-from langbot.pkg.api.http.context import ExecutionContext
-from langbot.pkg.cloud.entitlements import EntitlementResolver, EntitlementSnapshot
-from langbot.pkg.cloud.quotas import WorkspaceQuotaExceededError
-from langbot.pkg.plugin.connector import PluginRuntimeConnector
-from langbot_plugin.runtime.plugin.mgr import PluginInstallSource
-
-
-INSTANCE_UUID = 'cloud-instance'
-WORKSPACE_A = '11111111-1111-1111-1111-111111111111'
-WORKSPACE_B = '22222222-2222-2222-2222-222222222222'
-
-
-class _Provider:
-    async def get_workspace_entitlement(self, workspace_uuid: str) -> EntitlementSnapshot:
-        return EntitlementSnapshot(
-            instance_uuid=INSTANCE_UUID,
-            workspace_uuid=workspace_uuid,
-            entitlement_revision=1,
-            status='active',
-            not_before=0,
-            expires_at=4_102_444_800,
-            features={},
-            limits={'plugins.max': 3},
-        )
-
-
-class _Result:
-    def __init__(self, *, first=None, scalar=None) -> None:
-        self._first = first
-        self._scalar = scalar
-
-    def first(self):
-        return self._first
-
-    def scalar_one(self):
-        return self._scalar
-
-
-class _TenantUow:
-    def __init__(self, manager: '_Persistence', workspace_uuid: str) -> None:
-        self.manager = manager
-        self.workspace_uuid = workspace_uuid
-        self.lock = manager.locks[workspace_uuid]
-
-    async def __aenter__(self):
-        await self.lock.acquire()
-        return self
-
-    async def __aexit__(self, exc_type, exc, tb):
-        self.lock.release()
-
-    async def execute(self, statement):
-        sql = str(statement)
-        params = statement.compile().params
-        if isinstance(statement, sqlalchemy.sql.dml.Insert):
-            assert statement.table.name == 'plugin_settings'
-            key = (params['plugin_author'], params['plugin_name'])
-            self.manager.plugins[self.workspace_uuid][key] = dict(params)
-            return _Result()
-        if isinstance(statement, sqlalchemy.sql.dml.Update):
-            return _Result()
-        if 'FROM workspaces' in sql:
-            assert statement._for_update_arg is not None
-            self.manager.workspace_locks_seen += 1
-            return _Result(first=(self.workspace_uuid,))
-        if 'count(' in sql.lower() and 'FROM plugin_settings' in sql:
-            return _Result(scalar=len(self.manager.plugins[self.workspace_uuid]))
-        if 'FROM plugin_settings' in sql:
-            author = next(value for name, value in params.items() if 'plugin_author' in name)
-            name = next(value for param, value in params.items() if 'plugin_name' in param)
-            row = self.manager.plugins[self.workspace_uuid].get((author, name))
-            if row is None:
-                return _Result(first=None)
-            return _Result(
-                first=SimpleNamespace(
-                    installation_uuid=row['installation_uuid'],
-                    runtime_revision=row['runtime_revision'],
-                    artifact_digest=row['artifact_digest'],
-                    install_info=row['install_info'],
-                )
-            )
-        raise AssertionError(f'unexpected statement: {sql}')
-
-
-class _Persistence:
-    def __init__(self) -> None:
-        self.locks = defaultdict(asyncio.Lock)
-        self.plugins = defaultdict(dict)
-        self.workspace_locks_seen = 0
-
-    def tenant_uow(self, workspace_uuid: str) -> _TenantUow:
-        return _TenantUow(self, workspace_uuid)
-
-
-async def _connector(manager: _Persistence) -> PluginRuntimeConnector:
-    resolver = EntitlementResolver(INSTANCE_UUID, _Provider())
-    await resolver.reconcile_active_workspaces({WORKSPACE_A, WORKSPACE_B})
-    connector = object.__new__(PluginRuntimeConnector)
-    connector.ap = SimpleNamespace(entitlement_resolver=resolver, persistence_mgr=manager)
-    return connector
-
-
-def _context(workspace_uuid: str) -> ExecutionContext:
-    return ExecutionContext(
-        instance_uuid=INSTANCE_UUID,
-        workspace_uuid=workspace_uuid,
-        placement_generation=1,
-        entitlement_revision=1,
-    )
-
-
-@pytest.mark.asyncio
-async def test_cloud_plugin_quota_is_atomic_isolated_and_persists_across_connector_restart() -> None:
-    manager = _Persistence()
-    connector = await _connector(manager)
-
-    async def install(workspace_uuid: str, index: int):
-        return await connector._persist_installation_package(
-            _context(workspace_uuid),
-            plugin_author='test-author',
-            plugin_name=f'plugin-{index}',
-            install_source=PluginInstallSource.MARKETPLACE,
-            install_info={'author': 'test-author', 'name': f'plugin-{index}'},
-            artifact_digest=f'{index:064x}',
-        )
-
-    results = await asyncio.gather(
-        *(install(WORKSPACE_A, index) for index in range(10)),
-        *(install(WORKSPACE_B, index) for index in range(10)),
-        return_exceptions=True,
-    )
-
-    successes = [result for result in results if isinstance(result, tuple)]
-    failures = [result for result in results if isinstance(result, WorkspaceQuotaExceededError)]
-    assert len(successes) == 6
-    assert len(failures) == 14
-    assert len(manager.plugins[WORKSPACE_A]) == 3
-    assert len(manager.plugins[WORKSPACE_B]) == 3
-    assert manager.workspace_locks_seen == 20
-
-    restarted_connector = await _connector(manager)
-    with pytest.raises(WorkspaceQuotaExceededError, match=r'Maximum number of plugins \(3\) reached'):
-        await restarted_connector._persist_installation_package(
-            _context(WORKSPACE_A),
-            plugin_author='test-author',
-            plugin_name='after-restart',
-            install_source=PluginInstallSource.MARKETPLACE,
-            install_info={},
-            artifact_digest='f' * 64,
-        )
-    assert len(manager.plugins[WORKSPACE_A]) == 3
-
-    installed_name = next(iter(manager.plugins[WORKSPACE_A]))[1]
-
-    async def reinstall():
-        return await restarted_connector._persist_installation_package(
-            _context(WORKSPACE_A),
-            plugin_author='test-author',
-            plugin_name=installed_name,
-            install_source=PluginInstallSource.MARKETPLACE,
-            install_info={'author': 'test-author', 'name': installed_name, 'revision': 2},
-            artifact_digest='e' * 64,
-        )
-
-    reinstall_results = await asyncio.gather(reinstall(), reinstall())
-    assert all(result[2] is True for result in reinstall_results)
-
-    mixed_results = await asyncio.gather(
-        reinstall(),
-        restarted_connector._persist_installation_package(
-            _context(WORKSPACE_A),
-            plugin_author='test-author',
-            plugin_name='new-at-capacity',
-            install_source=PluginInstallSource.MARKETPLACE,
-            install_info={},
-            artifact_digest='d' * 64,
-        ),
-        return_exceptions=True,
-    )
-    assert isinstance(mixed_results[0], tuple)
-    assert isinstance(mixed_results[1], WorkspaceQuotaExceededError)
-    assert len(manager.plugins[WORKSPACE_A]) == 3

tests/unit_tests/plugin/test_connector_methods.py

@@ -612,8 +612,13 @@ class TestDisabledPluginEarlyReturns:
         mock_app.instance_config.data = {'plugin': {'enable': False}}
 
         connector = connector_module.PluginRuntimeConnector(mock_app, mock_disconnect)
+        execution_context = connector_module.ExecutionContext(
+            instance_uuid='instance-a',
+            workspace_uuid='workspace-a',
+            placement_generation=1,
+        )
 
-        result = await connector.get_debug_info()
+        result = await connector.get_debug_info(execution_context)
 
         assert result == {}
 

tests/unit_tests/plugin/test_connector_ping.py

@@ -282,12 +282,11 @@ def test_closed_deployment_selects_instance_scoped_shared_profile():
     assert connector.runtime_profile == 'shared'
 
 
-def test_external_runtime_control_headers_require_strong_secret(monkeypatch):
+def test_external_runtime_control_headers_are_empty_when_secret_is_unset(monkeypatch):
     monkeypatch.delenv(PLUGIN_RUNTIME_CONTROL_TOKEN_ENV, raising=False)
     connector = make_connector()
 
-    with pytest.raises(PluginRuntimeNotConnectedError, match=PLUGIN_RUNTIME_CONTROL_TOKEN_ENV):
-        connector._control_headers(allow_generate=False)
+    assert connector._control_headers(allow_generate=False) == {}
 
 
 def test_local_runtime_control_headers_generate_ephemeral_secret(monkeypatch):

tests/unit_tests/plugin/test_connector_reconcile.py

@@ -153,6 +153,31 @@ async def test_empty_projected_workspaces_do_not_retain_installation_sets():
     connector.handler.reconcile_plugin_installations.assert_awaited_once_with(())
 
 
+@pytest.mark.asyncio
+async def test_shared_reconcile_logs_workspace_installation_counts_and_elapsed_time():
+    binding_a = execution_binding('workspace-a')
+    binding_b = execution_binding('workspace-b')
+    setting_a = plugin_setting('01', 'a' * 64)
+    setting_b = plugin_setting('02', 'b' * 64)
+    connector = shared_connector(
+        [[binding_a, binding_b]],
+        {'workspace-a': [setting_a], 'workspace-b': [setting_b]},
+    )
+    connector.handler = runtime_handler()
+    await connector._prepare_connected_runtime()
+
+    matching_calls = [
+        call
+        for call in connector.ap.logger.info.call_args_list
+        if call.args
+        and call.args[0]
+        == 'Shared plugin runtime reconcile completed: workspaces=%d desired_installations=%d elapsed_seconds=%.3f'
+    ]
+    assert len(matching_calls) == 1
+    assert matching_calls[0].args[1:3] == (2, 2)
+    assert matching_calls[0].args[3] >= 0
+
+
 @pytest.mark.asyncio
 async def test_fresh_shared_runtime_cache_replays_persisted_local_package():
     package = b'local-lbpkg-bytes'

tests/unit_tests/plugin/test_connector_static.py

@@ -6,9 +6,10 @@ Tests cover:
 
 from __future__ import annotations
 
-import pytest
 from importlib import import_module
 
+import pytest
+
 
 def get_connector_module():
     """Lazy import to avoid circular import issues."""
@@ -60,3 +61,28 @@ def test_runtime_id_is_stable_across_core_restarts(monkeypatch):
     monkeypatch.setattr(connector.constants, 'instance_id', 'instance-a')
 
     assert connector.PluginRuntimeConnector._build_runtime_id() == 'instance-a:plugin-runtime'
+
+
+def test_runtime_connect_timeout_defaults_to_three_minutes():
+    connector = get_connector_module()
+    assert connector.PluginRuntimeConnector._runtime_connect_timeout({}) == 180.0
+
+
+def test_runtime_connect_timeout_reads_typed_plugin_config():
+    connector = get_connector_module()
+    assert connector.PluginRuntimeConnector._runtime_connect_timeout({'connect_timeout_seconds': 45.5}) == 45.5
+
+
+@pytest.mark.parametrize('value', [True, False, None, 0, -1, float('nan'), float('inf'), '180', object()])
+def test_runtime_connect_timeout_rejects_invalid_values(value):
+    connector = get_connector_module()
+    with pytest.raises(ValueError, match='plugin.connect_timeout_seconds'):
+        connector.PluginRuntimeConnector._runtime_connect_timeout({'connect_timeout_seconds': value})
+
+
+def test_runtime_connect_timeout_error_displays_actual_seconds():
+    connector = get_connector_module()
+
+    assert connector.PluginRuntimeConnector._runtime_connect_timeout_error(45.5) == (
+        'Plugin runtime did not become ready within 45.5 seconds'
+    )

tests/unit_tests/telemetry/test_heartbeat.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import json
+from datetime import datetime, timezone
 from types import SimpleNamespace
 
 import pytest
@@ -14,6 +15,12 @@ def get_heartbeat_module():
     return import_module('langbot.pkg.telemetry.heartbeat')
 
 
+def test_workspace_created_timestamp_treats_naive_database_values_as_utc():
+    heartbeat = get_heartbeat_module()
+    created_at = datetime(2026, 8, 4, 0, 0, 0)
+    assert heartbeat._workspace_created_timestamp(created_at) == 1785801600
+
+
 def make_app():
     ap = Mock()
     ap.instance_config = Mock()
@@ -57,13 +64,17 @@ def make_app():
 
 class TestBuildHeartbeatPayload:
     @pytest.mark.asyncio
-    async def test_payload_shape(self):
+    async def test_payload_shape(self, monkeypatch):
         heartbeat = get_heartbeat_module()
+        monkeypatch.setattr(heartbeat.constants, 'instance_id', 'instance-test')
         ap = make_app()
-        payload = await heartbeat.build_heartbeat_payload(ap)
+        payload = await heartbeat.build_heartbeat_payload(ap, workspace_uuid='workspace-a')
 
         assert payload['event_type'] == 'instance_heartbeat'
         assert payload['query_id'] == ''
+        assert payload['workspace_uuid'] == 'workspace-a'
+        assert payload['instance_id']
+        assert payload['workspace_create_ts'] == 0
         assert 'instance_create_ts' in payload
         assert 'timestamp' in payload
         f = payload['features']
@@ -86,7 +97,7 @@ class TestBuildHeartbeatPayload:
     @pytest.mark.asyncio
     async def test_payload_is_json_serializable(self):
         heartbeat = get_heartbeat_module()
-        payload = await heartbeat.build_heartbeat_payload(make_app())
+        payload = await heartbeat.build_heartbeat_payload(make_app(), workspace_uuid='workspace-a')
         json.dumps(payload)
 
     @pytest.mark.asyncio
@@ -94,12 +105,13 @@ class TestBuildHeartbeatPayload:
         heartbeat = get_heartbeat_module()
         ap = make_app()
         ap.persistence_mgr.execute_async = AsyncMock(side_effect=RuntimeError('db down'))
-        payload = await heartbeat.build_heartbeat_payload(ap)
+        payload = await heartbeat.build_heartbeat_payload(ap, workspace_uuid='workspace-a')
         assert payload['features']['pipeline_count'] == -1
 
     @pytest.mark.asyncio
-    async def test_cloud_counts_loaded_registries_without_tenant_sql(self):
+    async def test_cloud_counts_loaded_registries_without_tenant_sql(self, monkeypatch):
         heartbeat = get_heartbeat_module()
+        monkeypatch.setattr(heartbeat.constants, 'instance_id', 'instance-test')
         ap = make_app()
         ap.persistence_mgr.mode = SimpleNamespace(value='cloud_runtime')
         ap.persistence_mgr.execute_async = AsyncMock(
@@ -111,8 +123,10 @@ class TestBuildHeartbeatPayload:
                 ('instance-a', 'workspace-a', 'pipeline-b'): object(),
             },
         )
+        adapter_a = type('WorkspaceAAdapter', (), {})()
+        adapter_b = type('WorkspaceBAdapter', (), {})()
         ap.platform_mgr._bots_by_key = {
-            ('instance-a', 'workspace-a', 'bot-a'): object(),
+            ('instance-a', 'workspace-a', 'bot-a'): SimpleNamespace(enable=True, adapter=adapter_a),
         }
         ap.tool_mgr = SimpleNamespace(
             mcp_tool_loader=SimpleNamespace(
@@ -129,35 +143,54 @@ class TestBuildHeartbeatPayload:
         ap.plugin_connector._workspace_installations = {
             'workspace-a': {'plugin-a', 'plugin-b'},
         }
+        ap.skill_mgr._skills_by_scope = {
+            ('instance-a', 'workspace-a', 1): {'skill-a': {}, 'skill-b': {}},
+            ('instance-a', 'workspace-b', 1): {'skill-c': {}},
+        }
         ap.workspace_service.list_active_execution_bindings = AsyncMock(
-            return_value=[SimpleNamespace(workspace_uuid='workspace-a')],
+            return_value=[
+                SimpleNamespace(workspace_uuid='workspace-a', placement_generation=7),
+                SimpleNamespace(
+                    workspace_uuid='workspace-b',
+                    placement_generation=9,
+                    workspace_created_at=datetime(2026, 8, 4, tzinfo=timezone.utc),
+                ),
+            ],
+        )
+        ap.platform_mgr._bots_by_key[('instance-a', 'workspace-b', 'bot-b')] = SimpleNamespace(
+            enable=True, adapter=adapter_b
         )
 
-        payload = await heartbeat.build_heartbeat_payload(ap)
+        payloads = await heartbeat.build_heartbeat_payloads(ap)
 
-        features = payload['features']
-        assert features['pipeline_count'] == 2
-        assert features['mcp_server_count'] == 3
-        assert features['knowledge_base_count'] == 1
-        assert features['bot_count'] == 1
-        assert features['workspace_resources'] == [
-            {
-                'workspace_uuid': 'workspace-a',
-                'bot_count': 1,
-                'pipeline_count': 2,
-                'knowledge_base_count': 1,
-                'plugin_count': 2,
-                'mcp_server_count': 3,
-                'extension_count': 5,
-            }
-        ]
+        assert [payload['workspace_uuid'] for payload in payloads] == ['workspace-a', 'workspace-b']
+        assert all(payload['instance_id'] for payload in payloads)
+        assert payloads[0]['workspace_create_ts'] == 0
+        assert payloads[1]['workspace_create_ts'] == 1785801600
+        by_workspace = {payload['workspace_uuid']: payload['features'] for payload in payloads}
+        assert by_workspace['workspace-a']['pipeline_count'] == 2
+        assert by_workspace['workspace-a']['mcp_server_count'] == 3
+        assert by_workspace['workspace-a']['knowledge_base_count'] == 1
+        assert by_workspace['workspace-a']['bot_count'] == 1
+        assert by_workspace['workspace-a']['plugin_count'] == 2
+        assert by_workspace['workspace-a']['extension_count'] == 5
+        assert by_workspace['workspace-a']['skill_count'] == 2
+        assert by_workspace['workspace-a']['execution_generation'] == 7
+        assert by_workspace['workspace-a']['adapters'] == ['WorkspaceAAdapter']
+        assert by_workspace['workspace-b']['bot_count'] == 1
+        assert by_workspace['workspace-b']['pipeline_count'] == 0
+        assert by_workspace['workspace-b']['skill_count'] == 1
+        assert by_workspace['workspace-b']['execution_generation'] == 9
+        assert by_workspace['workspace-b']['adapters'] == ['WorkspaceBAdapter']
+        assert 'workspace_resources' not in by_workspace['workspace-a']
         ap.persistence_mgr.execute_async.assert_not_awaited()
+        ap.workspace_service.list_active_execution_bindings.assert_awaited_once()
 
     @pytest.mark.asyncio
     async def test_no_user_content_fields(self):
         """The heartbeat must never carry message content / credentials keys."""
         heartbeat = get_heartbeat_module()
-        payload = await heartbeat.build_heartbeat_payload(make_app())
+        payload = await heartbeat.build_heartbeat_payload(make_app(), workspace_uuid='workspace-a')
         flat = json.dumps(payload).lower()
         for forbidden in ('api_key', 'password', 'token', 'message_content'):
             assert forbidden not in flat

tests/unit_tests/telemetry/test_telemetry.py

@@ -569,6 +569,63 @@ class TestHTTPScenarios:
             await manager.send({'query_id': 'test'})
 
 
+class TestTelemetryManagedRuntimeAuthentication:
+    @pytest.mark.asyncio
+    async def test_send_includes_managed_runtime_token_header(self):
+        telemetry = get_telemetry_module()
+        mock_app = Mock()
+        mock_app.logger = Mock()
+        manager = telemetry.TelemetryManager(mock_app)
+        manager.telemetry_config = {'url': 'https://example.com'}
+        captured = {}
+
+        async def mock_post(url, json, headers):
+            captured['headers'] = headers
+            return Mock(status_code=200, text='', json=Mock(return_value={'code': 0}))
+
+        mock_client = Mock()
+        mock_client.post = mock_post
+        mock_client.__aenter__ = AsyncMock(return_value=mock_client)
+        mock_client.__aexit__ = AsyncMock(return_value=None)
+        with (
+            patch.dict('os.environ', {'LANGBOT_TELEMETRY_INGEST_TOKEN': 'managed-runtime-secret'}),
+            patch.object(httpx, 'AsyncClient', return_value=mock_client),
+        ):
+            await manager.send({'event_type': 'instance_heartbeat'})
+
+        assert captured['headers'] == {'X-LangBot-Telemetry-Token': 'managed-runtime-secret'}
+
+
+class TestAuthenticatedWorkspaceReporter:
+    @pytest.mark.asyncio
+    async def test_workspace_owner_access_token_is_sent_as_bearer(self):
+        telemetry = get_telemetry_module()
+        mock_app = Mock()
+        mock_app.logger = Mock()
+        mock_app.user_service = Mock()
+        mock_app.user_service.get_workspace_owner = AsyncMock(
+            return_value=Mock(user='owner@example.com', space_access_token='expired-token')
+        )
+        mock_app.space_service = Mock()
+        mock_app.space_service.get_valid_access_token = AsyncMock(return_value='refreshed-workspace-owner-token')
+        manager = telemetry.TelemetryManager(mock_app)
+        manager.telemetry_config = {'url': 'https://example.com'}
+
+        response = Mock(status_code=200, text='')
+        response.json = Mock(return_value={'code': 0})
+        mock_client = Mock()
+        mock_client.post = Mock(return_value=response)
+
+        with patch.object(httpx, 'AsyncClient', return_value=mock_client):
+            await manager.send({'query_id': 'q-1', 'workspace_uuid': 'workspace-1'})
+
+        mock_app.user_service.get_workspace_owner.assert_awaited_once_with('workspace-1')
+        mock_app.space_service.get_valid_access_token.assert_awaited_once_with('owner@example.com')
+        assert mock_client.post.call_args.kwargs['headers'] == {
+            'Authorization': 'Bearer refreshed-workspace-owner-token'
+        }
+
+
 class TestStartSendTask:
     """Tests for start_send_task() method."""
 

tests/unit_tests/telemetry/test_workspace_identity.py

@@ -0,0 +1,34 @@
+from __future__ import annotations
+
+import uuid
+from types import SimpleNamespace
+
+
+def test_standard_oss_instance_id_aligns_to_embedded_uuid():
+    from langbot.pkg.workspace.identity import workspace_uuid_from_instance_id
+
+    instance_uuid = 'a711d9e4-0953-443f-a0e9-7dd50193a79f'
+
+    assert workspace_uuid_from_instance_id(instance_uuid) == instance_uuid
+    assert workspace_uuid_from_instance_id(f'instance_{instance_uuid}') == instance_uuid
+
+
+def test_custom_legacy_instance_id_maps_to_stable_valid_uuid():
+    from langbot.pkg.workspace.identity import workspace_uuid_from_instance_id
+
+    first = workspace_uuid_from_instance_id('instance_migration_test')
+    second = workspace_uuid_from_instance_id('instance_migration_test')
+
+    assert first == second
+    assert str(uuid.UUID(first)) == first
+
+
+def test_query_telemetry_identity_reports_instance_and_workspace():
+    from langbot.pkg.telemetry.identity import workspace_identity
+
+    identity = workspace_identity(SimpleNamespace(workspace_uuid='workspace-a', instance_uuid='instance-a'))
+
+    assert identity == {
+        'instance_id': 'instance-a',
+        'workspace_uuid': 'workspace-a',
+    }

tests/unit_tests/vector/test_mgr.py

@@ -213,7 +213,7 @@ class TestVectorDBManagerInitialization:
                 mock_app,
                 connection_string='postgresql://user:pass@host:5432/langbot',
                 use_business_database=False,
-                allowed_dimensions=[384, 512, 768, 1024, 1536],
+                allowed_dimensions=[384, 512, 768, 1024, 1536, 3072],
             )
 
     def test_initialize_pgvector_with_individual_params(self):
@@ -251,7 +251,7 @@ class TestVectorDBManagerInitialization:
                 user='admin',
                 password='secret',
                 use_business_database=False,
-                allowed_dimensions=[384, 512, 768, 1024, 1536],
+                allowed_dimensions=[384, 512, 768, 1024, 1536, 3072],
             )
 
     def test_initialize_pgvector_defaults(self):
@@ -280,7 +280,7 @@ class TestVectorDBManagerInitialization:
                 user='postgres',
                 password='postgres',
                 use_business_database=False,
-                allowed_dimensions=[384, 512, 768, 1024, 1536],
+                allowed_dimensions=[384, 512, 768, 1024, 1536, 3072],
             )
 
     def test_initialize_pgvector_with_shared_business_database(self):

tests/unit_tests/workspace/test_workspace_collaboration.py

@@ -99,6 +99,7 @@ async def test_invitation_secret_is_hashed_and_acceptance_is_one_time(collaborat
     membership = await service.accept_invitation(created.token, account.uuid)
     assert membership.workspace_uuid == workspace.uuid
     assert membership.role == 'developer'
+    assert membership.source == 'local'
 
     with pytest.raises(InvitationUsedError):
         await service.accept_invitation(created.token, account.uuid)
@@ -203,20 +204,21 @@ async def test_last_owner_cannot_be_demoted(collaboration_context):
             second_membership,
         )
 
-    promoted = await service.update_member_role(
-        workspace.uuid,
-        second.uuid,
-        'owner',
-        owner_membership,
-    )
-    assert promoted.role == 'owner'
-    demoted = await service.update_member_role(
-        workspace.uuid,
-        owner_membership.account_uuid,
-        'admin',
-        owner_membership,
-    )
-    assert demoted.role == 'admin'
+    with pytest.raises(MembershipPermissionError, match='cannot be transferred'):
+        await service.update_member_role(
+            workspace.uuid,
+            second.uuid,
+            'owner',
+            owner_membership,
+        )
+
+    with pytest.raises(LastOwnerError):
+        await service.update_member_role(
+            workspace.uuid,
+            owner_membership.account_uuid,
+            'admin',
+            owner_membership,
+        )
 
 
 async def test_workspace_selector_requires_membership(collaboration_context):

tests/unit_tests/workspace/test_workspace_service.py

@@ -26,6 +26,7 @@ from langbot.pkg.workspace import (
     WorkspaceOwnerAlreadyExistsError,
     WorkspaceService,
 )
+from langbot.pkg.workspace.identity import workspace_uuid_from_instance_id
 from langbot.pkg.workspace.policy import CloudWorkspacePolicy
 
 
@@ -113,6 +114,7 @@ async def test_ensure_singleton_workspace_is_idempotent(workspace_test_context):
     first = await service.ensure_singleton_workspace()
     second = await service.ensure_singleton_workspace()
 
+    assert first.uuid == workspace_uuid_from_instance_id('instance_service_test')
     assert second.uuid == first.uuid
     async with session_factory() as session:
         assert await session.scalar(sqlalchemy.select(sqlalchemy.func.count()).select_from(Workspace)) == 1
@@ -151,6 +153,33 @@ async def test_initial_owner_cannot_be_claimed_by_another_account(workspace_test
         ).all()
         assert len(owners) == 1
         assert owners[0].account_uuid == first_account_uuid
+        assert owners[0].source == 'local'
+
+
+async def test_claim_initial_owner_reclassifies_existing_membership_as_local(workspace_test_context):
+    service, session_factory = workspace_test_context
+
+    async with session_factory() as session:
+        async with session.begin():
+            account_uuid = await _insert_account(session, 'reclaimed@example.com')
+            workspace = await service.ensure_singleton_workspace(session=session)
+            session.add(
+                WorkspaceMembership(
+                    uuid='44444444-4444-4444-8444-444444444444',
+                    workspace_uuid=workspace.uuid,
+                    account_uuid=account_uuid,
+                    role='viewer',
+                    status='removed',
+                    source='cloud_projection',
+                    projection_revision=4,
+                )
+            )
+
+    membership = await service.claim_initial_owner(account_uuid)
+
+    assert membership.role == 'owner'
+    assert membership.status == 'active'
+    assert membership.source == 'local'
 
 
 async def test_execution_binding_returns_persisted_generation(workspace_test_context):